Merge pull request 'docs: this describes the prod deployment now, not beta' (#11) from fix/docs-prod-forge-bases into main

Docs-only. NOTE: merged WITHOUT a completed granthi-review. The gate's webhook rejection was fixed today (item 52eea395), but reviews still fail open — diff fetch timeout, and shre-router returning prose instead of JSON (item 6d4e8819). Merged on explicit human approval, not on a green check.

README.md

@@ -1,8 +1,10 @@
 # granthi-sync v1.2
 
 The signup → download → link-folders → cloud product spine for the Granthi
-forge, tested against the BETA forge (granthi-beta.shre.ai). Python 3 stdlib +
-git CLI only — same portability heritage as the estate's `gitea_sync.py` mesh.
+forge. Developed and live-E2E-tested against the BETA forge
+(granthi-beta.shre.ai); **the deployed service now runs against the PRODUCTION
+forge** — see "Where this actually runs" below. Python 3 stdlib + git CLI only
+— same portability heritage as the estate's `gitea_sync.py` mesh.
 
 ```
    ┌──────────────┐   device flow    ┌─────────────────┐
@@ -13,7 +15,8 @@ git CLI only — same portability heritage as the estate's `gitea_sync.py` mesh.
    │              │  POST /v1/link {zitadel_access_token, device_name}
    │              │ ───────────────▶ ┌───────────────────────────────┐
    │              │ ◀─────────────── │ granthi-link  :3042           │
-   │              │  {login, token}  │ (granthi VPS, tailnet-only)   │
+   │              │  {login, token}  │ (granthi VPS; public via      │
+   │              │                  │  https://granthi-link.shre.ai)│
    │              │                  │  · userinfo validation        │
    │              │  POST /v1/repos  │  · ensure Gitea user (admin)  │
    │              │ ───────────────▶ │  · mint scoped user token     │
@@ -21,11 +24,28 @@ git CLI only — same portability heritage as the estate's `gitea_sync.py` mesh.
    │              │   git push/fetch (user token    │ admin API
    │              │   via credential helper)        ▼
    │              │ ───────────────▶ ┌───────────────────────────────┐
-   └──────────────┘                  │ BETA forge  :3041             │
-                                     │ granthi-beta.shre.ai          │
+   └──────────────┘                  │ PROD forge  :3040             │
+                                     │ granthi.shre.ai               │
                                      └───────────────────────────────┘
 ```
 
+## Where this actually runs
+
+Verified on the granthi VPS (`[email protected]`) on 2026-08-30, because
+this file previously described the beta tier long after the deployment moved:
+
+| | value |
+|---|---|
+| service | `/opt/granthi-link/`, `granthi-link.service`, `systemctl is-active` → `active` |
+| public entry | `https://granthi-link.shre.ai` — `/health` → 200. `/` → 404 is **no root route**, not an outage |
+| `gitea_base` | `http://127.0.0.1:3040` (container `gitea-central-gitea-1`) |
+| `public_gitea_base` | `https://granthi.shre.ai` |
+| rate limiting | `trust_forwarded_for: true`, `trusted_proxies: ["100.107.37.98/32"]`, no `rules` key → falls back to `DEFAULT_RATE_RULES` |
+| deployed version | `/health` reports **1.1.0** while this repo is **v1.2** — the running service lags `main` |
+
+So a new computer that follows Quickstart lands on the **production** forge.
+Beta remains where changes are proven before they reach it.
+
 ## Quickstart (invited user)
 
 You need a shre-id account — an operator creates it; there is no open signup
@@ -194,13 +214,16 @@ So:
   **64 KB** (413 beyond; missing `Content-Length` → 411, invalid → 400).
 * `POST /v1/repos {token, name, private}` → creates the user repo with the
   USER token; clone/html URLs are rebased onto `public_gitea_base` because the
-  container `ROOT_URL` (https://granthi-beta.shre.ai) does not resolve for
-  tailnet-only clients.
+  container `ROOT_URL` does not necessarily resolve for the client that asked
+  (it was a tailnet-only address on beta; on prod the rebase keeps clone URLs
+  on `https://granthi.shre.ai` rather than the container's own view).
 
 Deployment: `/opt/granthi-link/{granthi_link.py,config.json,state.json}` +
 systemd unit `granthi-link.service`; binds `127.0.0.1:3042` **and**
-`100.111.127.127:3042` (tailnet). **Not publicly exposed** — see promotion
-window. The service **refuses to start** (exit 2) unless `config.json` is
+`100.111.127.127:3042` (tailnet), and is **publicly reachable** at
+`https://granthi-link.shre.ai` through the `pulse-granthi-edge` cloudflared
+tunnel (done 2026-08-23; re-verified 2026-08-30). The service **refuses to
+start** (exit 2) unless `config.json` is
 mode 0600/0400 and owned by the user it runs as — the config carries the
 forge admin password, so permissive perms fail closed, not open.
 
@@ -412,7 +435,7 @@ deleted it again, `DELETE …/tokens/{id}` returning 204 under basic auth):
 
 ## Tests
 
-* `python3 -m unittest discover -s tests` — 184 tests. The v1.2 additions
+* `python3 -m unittest discover -s tests` — 198 tests. The v1.2 additions
   cover: a snapshot capturing uncommitted work while HEAD, the index and the
   working tree stay byte-identical; snapshots landing outside `refs/heads`;
   an unchanged tree not being re-pushed; a diverged folder still being backed
@@ -541,7 +564,48 @@ vault keys must exist, with the `shre-cred request` line to supply them.
 Per-repo `mode` in the manifest overrides the default, so a documents folder
 can be declared `mirror` while everything else stays on the safe `snapshot`.
 
-## Next phase — invites and per-repo access (designed, not built)
+## Sharing: grants and invites
+
+Two commands, because there are two situations.
+
+**They already have an account** — share directly:
+
+    granthi-sync share bob --repo notes                 # write by default
+    granthi-sync share bob --repo notes --permission read
+    granthi-sync share bob --repo notes --revoke
+    granthi-sync shared --repo notes                    # who can see it
+
+**They do not have an account yet** — invite them:
+
+    granthi-sync invite carol@example.com --repo notes --repo reports
+
+An invite creates **nothing**: no account, no token, no collaborator row. The
+grant is held against their email and applied the first time they run
+`granthi-sync link`. An invite that is never accepted leaves nothing behind.
+
+Three properties worth keeping:
+
+* **The forge decides who may share.** Before recording anything, the service
+  asks Gitea whether the caller can administer that repo (listing
+  collaborators requires repo admin, so a 200 is Gitea's own answer). Deciding
+  it here from the repo name would be a second opinion about someone else's
+  authorisation — and the wrong one the first time a repo is transferred.
+* **An unverified email collects nothing.** The address is the only thing
+  tying a promise to a person, so an invite is applied only when the IdP says
+  the address is verified. The invite stays pending rather than being consumed.
+* **A failed grant never blocks a sign-in.** If a promised repo has since been
+  deleted, the person still links successfully and the failure is audited.
+  Someone must not be locked out of their own account by somebody else's
+  stale invite.
+
+Applying an invite uses the admin credential deliberately: the inviter
+authorised it when they issued it, and their session is long gone by the time
+it is redeemed. Everything else — sharing, unsharing, listing — runs on the
+caller's own token, which is why granting needs no elevated rights at all
+(verified against the live forge: a repo owner's scoped token adds and removes
+collaborators, HTTP 204).
+
+## Superseded design note — invites and per-repo access
 
 Today `/v1/link` creates an account and every folder becomes a private repo
 under it. What is missing is the multi-person case: an existing account
@@ -579,10 +643,12 @@ Shape this should take, so the next session does not re-litigate it:
    `https://granthi-link.shre.ai` (second-level, see above), origin stays
    tailnet-only, `trust_forwarded_for` on with the tunnel as the only
    trusted proxy.
-2. **Swap forge base URLs** in `/opt/granthi-link/config.json`:
-   `gitea_base` → prod forge, `public_gitea_base` →
-   `https://granthi.shre.ai`; the client default server URL moves to the
-   public endpoint.
+2. ~~**Swap forge base URLs** in `/opt/granthi-link/config.json`~~
+   **DONE — verified live 2026-08-30**: the deployed config reads
+   `gitea_base: http://127.0.0.1:3040` and
+   `public_gitea_base: https://granthi.shre.ai`, and the client default
+   server is already the public endpoint. Linking a new device therefore
+   creates the account on **prod**.
 3. The `granthi-web` OIDC app already lists the prod callback; the device
    app is host-independent. Rotate the beta admin token/password out of the
    config when pointing at prod (prod forge is READ-ONLY to this estate —

SETUP.md

@@ -0,0 +1,163 @@
+# Setting up Granthi on a new computer
+
+Everything here is one command at a time. You need `git` and `python3`
+(3.9 or newer). There is nothing to install — no package, no build.
+
+## Read this first: what you will see when you sign in
+
+Signing in does **not** hand you an account you already had somewhere else.
+Granthi looks up your shre-id identity, and:
+
+* if it has seen you before, you get the same forge account as last time;
+* if it has not, and the login it would use is free, it makes you a **new,
+  empty account**.
+
+So the first time you sign in on a new computer, `granthi-sync list` may show
+**nothing**. That is not a failure. It means you are a new account and nobody
+has given you access to anything yet.
+
+Granthi refuses to hand you an existing account just because the name lines
+up. It will only join you to one if that account's email is the same as your
+verified sign-in email. That rule is what stops somebody else claiming your
+files, so it is worth the small surprise on day one.
+
+**Two ways to end up with your files, and it is worth choosing on purpose:**
+
+| | what happens | when to pick it |
+|---|---|---|
+| **Fresh account** (recommended) | You sign in, get a new empty account, and somebody shares the repos you need with you | Your everyday laptop. The token stored on the machine can only do what that account can do. |
+| **Join an existing account** | An operator sets that account's email to your verified sign-in email first; then signing in joins you to it and you see everything it owns straight away | You want one identity everywhere and accept that the laptop then carries whatever that account can do |
+
+The fresh account is recommended for a plain reason: the token `link` saves on
+the machine acts as that account. If you join an admin account, every computer
+you sync from is carrying admin.
+
+## 1. Get the program
+
+    git clone https://granthi.shre.ai/nirpa/granthi-sync.git
+    cd granthi-sync
+
+No account is needed for this step.
+
+## 2. Sign in
+
+    ./bin/granthi-sync link
+
+It prints a web address and a short code. Open the address, type the code, and
+approve — you have five minutes. When it finishes, your account exists and a
+key is saved at `~/.granthi-sync/config.json`, readable only by you. You never
+see or type a forge password.
+
+If the code runs out, **nothing was created** — no account, no key, no
+half-finished state. Run `link` again.
+
+You do not need to be on any private network. The client talks to
+`https://granthi-link.shre.ai`.
+
+## 3. See what you have
+
+    ./bin/granthi-sync list
+    ./bin/granthi-sync list work-          # only names containing "work-"
+
+This list comes from the forge, so it is exactly what you are allowed to see —
+not a guess made on this computer.
+
+If it is empty, see the note at the top: you are a new account. Ask whoever
+owns the repos to run `granthi-sync share <your-login> --repo <name>`, or to
+add you to the right organisation.
+
+## 4. Bring folders down
+
+    ./bin/granthi-sync get notes                  # by name
+    ./bin/granthi-sync get Nirlabinc/notes        # when two teams both have a "notes"
+    ./bin/granthi-sync get --all --into ~/work    # everything you are allowed to see
+
+If a plain name matches two repos, Granthi stops and shows you both rather
+than guessing which one you meant.
+
+## 5. Or send a folder up
+
+    ./bin/granthi-sync add ~/Documents/notes
+
+It becomes a private repo. Two things happen for your protection:
+
+* a starter `.gitignore` is written if the folder has none, covering `.env`,
+  `*.key`, `*.pem`, `id_rsa` and similar — those will **not** be synced;
+* a folder holding more than 20,000 files or 512 MB is refused unless you add
+  `--force`, so you cannot upload your whole disk by accident.
+
+## 6. Keep it syncing
+
+Run it in a terminal:
+
+    ./bin/granthi-sync watch
+
+Or leave it running in the background on a Mac:
+
+    cp client/launchd/ai.granthi.sync.plist ~/Library/LaunchAgents/
+
+Open that copied file and replace `/PATH/TO/granthi-sync` with the real path
+to this folder. Then:
+
+    launchctl bootstrap gui/$UID ~/Library/LaunchAgents/ai.granthi.sync.plist
+
+Its log is `/tmp/granthi-sync.log`.
+
+### What syncing does to your folder
+
+It depends on what the folder was:
+
+* **A folder that was already a git project** — Granthi **never commits for
+  you**. Your history stays yours. Work you have not committed is copied to
+  the cloud with a timestamp so it is not only on this laptop.
+* **A plain folder** — Granthi saves changes for you as it goes, and each save
+  is a point you can go back to.
+
+Either way, if your copy and the cloud copy have both moved on, Granthi
+**stops and tells you**. It will never merge or overwrite your work to make
+the conflict go away.
+
+## 7. Going back to an earlier version
+
+    ./bin/granthi-sync snapshots ~/Documents/notes
+    ./bin/granthi-sync restore  ~/Documents/notes --at 20260823T142530Z
+
+Restore writes to a **new folder**. It never overwrites what you have now —
+if you are restoring, you have enough problems already.
+
+You can restore from a computer that never had the folder, including one
+replacing a laptop that is gone.
+
+## 8. Sharing
+
+    ./bin/granthi-sync share bob --repo notes                 # they have an account
+    ./bin/granthi-sync share bob --repo notes --revoke
+    ./bin/granthi-sync shared --repo notes                    # who can see it
+    ./bin/granthi-sync invite carol@example.com --repo notes  # they do not yet
+
+An invite creates nothing until it is accepted. If the person never signs in,
+nothing was ever made for them.
+
+## 9. Keeping track of your computers
+
+    ./bin/granthi-sync devices     # every computer signed in to your account
+    ./bin/granthi-sync logout      # sign this computer out
+    ./bin/granthi-sync logout --device <id>   # sign out a lost one
+    ./bin/granthi-sync activity    # sign-ins and sign-outs, with times
+
+Signing a computer out takes effect **at the server**, immediately. The key on
+that machine stops working on its very next attempt — it is not a setting that
+machine could ignore. That is what makes it useful for a laptop you have lost.
+
+Your folders are left alone when you sign out; only the key is removed.
+
+## If something looks wrong
+
+* **`list` is empty** — you are a new account, nobody has shared anything yet.
+* **"Repository not found"** — you may not have access to that repo; run
+  `list` to see what you do have.
+* **"this device's access has been revoked"** — someone signed this computer
+  out. Run `link` again.
+* **A folder shows DIVERGED** — your copy and the cloud copy both changed.
+  Granthi is waiting for you on purpose. Nothing is lost; sort it out in git
+  or on the forge web page.

client/granthi_sync_client.py

@@ -104,6 +104,10 @@ FORGE_MAX_PAGES = 40  # 2000 repos; a guard against an unbounded paging loop
 # Under refs/heads a machine taking a snapshot every 30s would bury the
 # user's real branches.
 BACKUP_NS = "refs/granthi-backup"
+_SNAPSHOT_IDENT = {"GIT_AUTHOR_NAME": "granthi-sync",
+                   "GIT_AUTHOR_EMAIL": "[email protected]",
+                   "GIT_COMMITTER_NAME": "granthi-sync",
+                   "GIT_COMMITTER_EMAIL": "[email protected]"}
 SNAPSHOT_TS_FMT = "%Y%m%dT%H%M%SZ"
 
 # Retention. Unbounded snapshots are a disk leak with no way to use them, so
@@ -301,15 +305,33 @@ def build_snapshot(folder):
         args = ["commit-tree", tree, "-m", f"granthi snapshot: {ts}"]
         if head:
             args += ["-p", head]
+        # The worktree tree alone loses STAGED-ONLY work. Stage a hunk, edit the
+        # file further, lose the laptop, and the snapshot holds only the later
+        # worktree version -- the carefully staged one is gone. git itself keeps
+        # index and worktree as separate states, so the backup must too.
+        # The user's real index is read WITHOUT touching it, and when it differs
+        # from both HEAD and the worktree it rides along as a second parent, so
+        # it is reachable from the snapshot. (codex review, P2.)
+        rc_idx, index_tree = git(folder, "write-tree", check=False)
+        index_tree = index_tree.strip()
+        if rc_idx == 0 and index_tree and index_tree != tree:
+            head_tree_now = ""
+            if head:
+                head_tree_now = git(folder, "rev-parse", f"{head}^{{tree}}")[1].strip()
+            if index_tree != head_tree_now:
+                icommit_args = ["commit-tree", index_tree, "-m",
+                                f"granthi snapshot (staged): {ts}"]
+                if head:
+                    icommit_args += ["-p", head]
+                rc_ic, icommit = git(folder, *icommit_args, check=False,
+                                     env=_SNAPSHOT_IDENT)
+                if rc_ic == 0 and icommit.strip():
+                    args += ["-p", icommit.strip()]
         # Snapshots are parented on HEAD and nothing else -- deliberately NOT
         # chained to the previous snapshot. Chaining would keep every old
         # snapshot reachable from the newest one, so pruning a ref would free
         # nothing and retention would be decorative.
-        _, commit = git(folder, *args,
-                        env={"GIT_AUTHOR_NAME": "granthi-sync",
-                             "GIT_AUTHOR_EMAIL": "[email protected]",
-                             "GIT_COMMITTER_NAME": "granthi-sync",
-                             "GIT_COMMITTER_EMAIL": "[email protected]"})
+        _, commit = git(folder, *args, env=_SNAPSHOT_IDENT)
         return commit.strip(), tree
     finally:
         if os.path.exists(tmp_index):
@@ -599,8 +621,13 @@ def device_flow():
         form={"client_id": DEVICE_CLIENT_ID, "scope": DEVICE_SCOPE})
     if status != 200:
         raise SystemExit(f"device authorization failed (HTTP {status}): {resp}")
-    print(f"\nTo link this device, open:\n\n  {resp.get('verification_uri_complete') or resp.get('verification_uri')}\n")
-    print(f"and enter code: {resp['user_code']}\n")
+    # flush=True is not cosmetic. Python buffers stdout when it is not a
+    # terminal, so `granthi-sync link | tee setup.log`, a wrapper script, or
+    # anything capturing output shows NOTHING while the code silently expires
+    # five minutes later. Hit for real on 2026-08-23 driving a first sign-in.
+    print(f"\nTo link this device, open:\n\n  {resp.get('verification_uri_complete') or resp.get('verification_uri')}\n",
+          flush=True)
+    print(f"and enter code: {resp['user_code']}\n", flush=True)
     interval = int(resp.get("interval", 5))
     deadline = time.time() + int(resp.get("expires_in", 300))
     while time.time() < deadline:
@@ -833,6 +860,59 @@ def clone_one(cfg, full_name, dest, mode=None):
     return meta
 
 
+def resolve_granted(cfg, want, repos=None, truncated=False):
+    """Turn what the user typed into the repo the forge actually grants them.
+
+    `get notes` used to mean `<your-login>/notes` and nothing else. On a real
+    account that is wrong for almost everything: of 172 repos granted to this
+    estate's own admin, 157 are owned by an ORG, so the bare name failed for
+    91% of what the user could see and the error said "Repository not found"
+    -- which reads like a permissions problem rather than a naming one.
+
+    So a bare name is resolved against the granted list, which is the forge's
+    own answer about what this account may have. An owner-qualified name is
+    taken as given. An ambiguous bare name is REFUSED with the candidates
+    rather than guessed: picking one of two repos called `notes` owned by
+    different teams is not a guess worth making for someone.
+    """
+    if "/" in want:
+        return parse_repo_arg(want, cfg["login"])
+    if repos is None:
+        try:
+            repos, truncated = list_repos(cfg)
+        except (SystemExit, ValueError, OSError) as e:
+            # Best effort. If the listing is unreachable, a name the user
+            # typed in full must still clone, and a bare name should degrade
+            # to their own namespace rather than refusing outright -- the
+            # clone that follows reports the real problem precisely.
+            log(f"could not read your repo list ({e}); assuming "
+                f"{cfg['login']}/{want}")
+            return parse_repo_arg(want, cfg["login"])
+    matches = [r.get("full_name") for r in repos
+               if (r.get("full_name") or "").rsplit("/", 1)[-1] == want]
+    if len(matches) == 1:
+        return matches[0]
+    if len(matches) > 1:
+        listed = "\n".join(f"  {m}" for m in sorted(matches))
+        raise SystemExit(
+            f"{want!r} is ambiguous — {len(matches)} repos you can see have "
+            f"that name:\n{listed}\nRe-run with the owner, e.g. "
+            f"granthi-sync get {sorted(matches)[0]}")
+    # Nothing granted by that name. If the listing was TRUNCATED the answer is
+    # unknown rather than absent -- falling back to <login>/<name> could clone a
+    # different repo that happens to exist under your own account. Refuse and
+    # ask for the owner. (codex review, P3.)
+    if truncated:
+        raise SystemExit(
+            f"could not confirm {want!r}: your repo list was truncated at "
+            f"{FORGE_MAX_PAGES * FORGE_PAGE_LIMIT} repos, so a match may exist "
+            f"beyond it. Re-run with the owner, e.g. "
+            f"granthi-sync get <owner>/{want}")
+    # Otherwise the listing was complete and simply has no such repo; name a
+    # concrete one so the clone error is precise.
+    return parse_repo_arg(want, cfg["login"])
+
+
 def cmd_get(args):
     cfg = require_linked(load_config())
     device_id(cfg)
@@ -840,7 +920,7 @@ def cmd_get(args):
         return _get_all(cfg, args)
     if not args.repo:
         raise SystemExit("give a repo name, or --all")
-    full_name = parse_repo_arg(args.repo, cfg["login"])
+    full_name = resolve_granted(cfg, args.repo)
     name = full_name.rsplit("/", 1)[-1]
     clone_one(cfg, full_name, os.path.abspath(args.into or name), args.mode)
     return 0
@@ -1293,12 +1373,18 @@ def cmd_bootstrap(args):
     into = os.path.abspath(args.into or ".")
     print(f"workspace: {name}\n")
 
-    granted = {r.get("full_name") for r in list_repos(cfg)[0]}
-    granted |= {(r.get("full_name") or "").rsplit("/", 1)[-1] for r in []}
+    granted_repos, granted_truncated = list_repos(cfg)
+    granted = {r.get("full_name") for r in granted_repos}
     pulled = skipped = denied = 0
     for repo in ws["repos"]:
         want = repo["name"]
-        full = want if "/" in want else f"{cfg['login']}/{want}"
+        try:
+            full = resolve_granted(cfg, want, repos=granted_repos,
+                                   truncated=granted_truncated)
+        except SystemExit as e:
+            log(f"AMBIGUOUS {want}: {e}")
+            denied += 1
+            continue
         if full not in granted:
             # The forge decides. A manifest asking for a repo this account was
             # not granted is a permissions answer, not an error to route round.
@@ -1339,6 +1425,76 @@ def cmd_bootstrap(args):
     return 0
 
 
+def _post(cfg, path, body):
+    status, resp = http_json("POST", f"{cfg['server']}{path}", body=body)
+    if status == 401:
+        raise SystemExit("this device's access has been revoked -- "
+                         "run: granthi-sync link")
+    return status, resp
+
+
+def cmd_share(args):
+    """Give someone who already has an account access to one of your repos."""
+    cfg = require_linked(load_config())
+    status, resp = _post(cfg, "/v1/grants",
+                         {"token": cfg["token"],
+                          "action": "remove" if args.revoke else "add",
+                          "repo": args.repo, "login": args.who,
+                          "permission": args.permission})
+    if status == 403:
+        raise SystemExit(f"you cannot administer {args.repo} -- only the "
+                         f"owner can share it")
+    if status != 200:
+        raise SystemExit(f"share failed (HTTP {status}): {resp}")
+    verb = "no longer shared with" if args.revoke else \
+        f"shared with ({resp.get('permission')})"
+    log(f"{resp.get('repo')} {verb} {resp.get('login')}")
+    return 0
+
+
+def cmd_shared(args):
+    """Who can see one of your repos."""
+    cfg = require_linked(load_config())
+    status, resp = _post(cfg, "/v1/grants",
+                         {"token": cfg["token"], "action": "list",
+                          "repo": args.repo})
+    if status == 403:
+        raise SystemExit(f"you cannot administer {args.repo}")
+    if status != 200:
+        raise SystemExit(f"could not list access (HTTP {status}): {resp}")
+    people = resp.get("collaborators") or []
+    if not people:
+        print(f"{resp.get('repo')}: nobody else has access")
+        return 0
+    print(f"{resp.get('repo')} is shared with:")
+    for p in people:
+        print(f"  {p}")
+    return 0
+
+
+def cmd_invite(args):
+    """Promise access to someone who has no account yet.
+
+    Nothing is created for them now. The grant is held against their email and
+    applied the first time they sign in -- so an invite that is never accepted
+    leaves nothing behind.
+    """
+    cfg = require_linked(load_config())
+    repos = [{"name": r, "permission": args.permission} for r in args.repo]
+    status, resp = _post(cfg, "/v1/invite",
+                         {"token": cfg["token"], "email": args.email,
+                          "repos": repos})
+    if status == 403:
+        raise SystemExit(str(resp.get("error") or "you cannot share that repo"))
+    if status != 200:
+        raise SystemExit(f"invite failed (HTTP {status}): {resp}")
+    log(f"invited {resp['invited']} to "
+        f"{', '.join(g['repo'] for g in resp['repos'])}")
+    log("they get access the first time they run `granthi-sync link` with "
+        "that email -- until then nothing exists for them")
+    return 0
+
+
 def _folder_meta(cfg, folder):
     path = os.path.abspath(folder)
     meta = cfg.get("folders", {}).get(path)
@@ -1552,6 +1708,28 @@ def main(argv=None):
     sp.add_argument("--dry-run", action="store_true")
     sp.set_defaults(fn=cmd_bootstrap)
 
+    sp = sub.add_parser("share", help="share one of your repos with someone")
+    sp.add_argument("who", help="their forge login")
+    sp.add_argument("--repo", required=True)
+    sp.add_argument("--permission", default="write",
+                    choices=("read", "write", "admin"))
+    sp.add_argument("--revoke", action="store_true",
+                    help="take the access away again")
+    sp.set_defaults(fn=cmd_share)
+
+    sp = sub.add_parser("shared", help="who can see one of your repos")
+    sp.add_argument("--repo", required=True)
+    sp.set_defaults(fn=cmd_shared)
+
+    sp = sub.add_parser("invite",
+                        help="invite someone who has no account yet")
+    sp.add_argument("email")
+    sp.add_argument("--repo", action="append", required=True,
+                    help="repeatable")
+    sp.add_argument("--permission", default="write",
+                    choices=("read", "write", "admin"))
+    sp.set_defaults(fn=cmd_invite)
+
     sp = sub.add_parser("status", help="show linked folders")
     sp.set_defaults(fn=cmd_status)
 

server/config.example.json

@@ -1,15 +1,17 @@
 {
-  "gitea_base": "http://127.0.0.1:3041",
-  "public_gitea_base": "http://100.111.127.127:3041",
+  "_comment_forge": "Values below mirror the LIVE prod deployment (verified 2026-08-30). For the beta tier use gitea_base http://127.0.0.1:3041, public_gitea_base https://granthi-beta.shre.ai, container gitea-beta-gitea-1, creds /opt/gitea-beta/.admin-creds.",
+  "gitea_base": "http://127.0.0.1:3040",
+  "public_gitea_base": "https://granthi.shre.ai",
   "zitadel_userinfo": "https://id.shre.ai/oidc/v1/userinfo",
-  "admin_token": "MINT-VIA: docker exec -u git gitea-beta-gitea-1 gitea admin user generate-access-token --username nirpa --scopes write:admin,write:user,write:repository --raw",
+  "admin_token": "MINT-VIA: docker exec -u git gitea-central-gitea-1 gitea admin user generate-access-token --username nirpa --scopes write:admin,write:user,write:repository --raw",
   "admin_login": "nirpa",
-  "admin_password": "FROM /opt/gitea-beta/.admin-creds (required: Gitea 1.27 token minting only works via basic auth + Sudo header)",
+  "admin_password": "FROM shre-cred: superadmin/granthi-link/granthi-prod-forge-admin (required: Gitea 1.27 token minting only works via basic auth + Sudo header). NEVER paste this into chat or a shell history.",
   "binds": [["127.0.0.1", 3042], ["100.111.127.127", 3042]],
   "test_mode": false,
   "rate_limit": {
     "enabled": true,
-    "trust_forwarded_for": false,
+    "trust_forwarded_for": true,
+    "trusted_proxies": ["100.107.37.98/32"],
     "rules": {"/v1/link": [5, 3600], "/v1/repos": [60, 3600]}
   }
 }

server/granthi_link.py

@@ -80,6 +80,9 @@ TEST_MODE_ENV = "GRANTHI_LINK_ALLOW_TEST_MODE"
 
 # Sentinel: create_user hit a 409 (someone else created the login first).
 USER_CREATE_CONFLICT = object()
+# Sentinel: the address already belongs to another forge account, which is a
+# 409 the caller can act on -- not a 502 that reads like the service is down.
+EMAIL_IN_USE = object()
 
 LOGIN_SAFE = re.compile(r"[^a-zA-Z0-9._-]+")
 
@@ -90,6 +93,10 @@ DEFAULT_RATE_RULES = {"/v1/link": (5, 3600), "/v1/repos": (60, 3600),
                       # Reads are cheap but still authenticated work.
                       "/v1/devices": (120, 3600),
                       "/v1/audit": (120, 3600),
+                      # Sharing is an ordinary act; inviting reaches a person
+                      # who does not exist yet, so it is the tighter of the two.
+                      "/v1/grants": (120, 3600),
+                      "/v1/invite": (60, 3600),
                       # Revocation is a safety action, so its limit is set
                       # high rather than tight. It is NOT 0: in this limiter
                       # a limit of 0 DISABLES the endpoint outright (see
@@ -288,6 +295,14 @@ def check_config_perms(path, euid=None):
 # Identity map: zitadel sub -> gitea login (JSON, 0600, atomic writes)
 # --------------------------------------------------------------------------
 
+def _email_in_use_message(login, userinfo):
+    email = userinfo.get("email") or "your address"
+    return (f"cannot create the account '{login}': {email} already belongs to "
+            f"a different account on this forge. If that other account is "
+            f"yours, an operator must bind your identity to it; if it is not, "
+            f"use a different address.")
+
+
 class IdentityStore:
     """Persistent map of Zitadel `sub` -> Gitea login binding records.
 
@@ -368,6 +383,59 @@ class IdentityStore:
                 return sub, dict(ident.get("devices") or {})
         return None, {}
 
+    # -- pending invites ---------------------------------------------------
+    # Keyed by VERIFIED email. An invite is a promise of access made before the
+    # person has a forge account; it is applied the first time they link. The
+    # key must be an identity the IdP vouches for, or anyone could claim
+    # someone else's invite by asserting their address.
+
+    def add_invite(self, email, grants, invited_by):
+        data = self._load()
+        book = data.setdefault("invites", {})
+        entry = book.setdefault(email.strip().lower(), [])
+        for g in grants:
+            # Re-inviting the same repo updates the permission rather than
+            # stacking duplicates that would be applied twice.
+            entry[:] = [e for e in entry if e["repo"] != g["repo"]]
+            entry.append({"repo": g["repo"], "permission": g["permission"],
+                          "invited_by": invited_by,
+                          "invited_at": datetime.now(timezone.utc).isoformat(
+                              timespec="seconds")})
+        self._write(data)
+
+    def take_invites(self, email):
+        """Read AND clear the invites for an email, atomically under the
+        caller's lock. Returns [] when there are none."""
+        if not email:
+            return []
+        data = self._load()
+        book = data.get("invites") or {}
+        pending = book.pop(email.strip().lower(), [])
+        if pending:
+            self._write(data)
+        return pending
+
+    def consume_invite(self, email, repo):
+        """Drop ONE applied grant, leaving any that failed still pending.
+
+        The whole-list `take_invites` is what made a transient forge error
+        permanent; this removes only what actually landed.
+        """
+        data = self._load()
+        book = data.get("invites") or {}
+        key = (email or "").strip().lower()
+        entry = [e for e in book.get(key, []) if e.get("repo") != repo]
+        if entry:
+            book[key] = entry
+        else:
+            book.pop(key, None)
+        data["invites"] = book
+        self._write(data)
+
+    def peek_invites(self, email):
+        book = self._load().get("invites") or {}
+        return list(book.get((email or "").strip().lower(), []))
+
     def mark_revoked(self, sub, device_id, when):
         data = self._load()
         ident = data["identities"].get(str(sub)) or {}
@@ -620,6 +688,14 @@ class LinkService:
             headers=self._admin_hdr(), body=body)
         if status == 409:
             return USER_CREATE_CONFLICT
+        if status == 422 and "e-mail already in use" in str(resp).lower():
+            # The address belongs to a DIFFERENT forge account. Reported as its
+            # own case because the generic path turns it into a bare 502, and
+            # "502" sends the person looking for an outage when the real answer
+            # is "that address is already somebody's account here". Hit live on
+            # 2026-08-23: an operator moved an email onto another account and
+            # the next sign-in failed with nothing but the number.
+            return EMAIL_IN_USE
         if status != 201:
             return f"gitea admin user create failed (HTTP {status}): {resp}"
         return None
@@ -676,6 +752,8 @@ class LinkService:
                                  "was not created by this service; refusing "
                                  "to re-create"}, None
                 err = self.create_user(login, userinfo)
+                if err is EMAIL_IN_USE:
+                    return 409, {"error": _email_in_use_message(login, userinfo)}, None
                 if err and err is not USER_CREATE_CONFLICT:
                     return 502, {"error": err}, None
                 LOG.info("re-created service-managed gitea user %s", login)
@@ -701,6 +779,8 @@ class LinkService:
                     return 409, {"error": "login exists and is not linked "
                                           "to this identity"}, None
                 LOG.info("user %s created concurrently; continuing", login)
+            elif err is EMAIL_IN_USE:
+                return 409, {"error": _email_in_use_message(login, userinfo)}, None
             elif err:
                 return 502, {"error": err}, None
             else:
@@ -767,12 +847,16 @@ class LinkService:
                 "linked_at": datetime.now(timezone.utc).isoformat(
                     timespec="seconds"),
             })
+        granted = self.apply_invites(login, userinfo, client_ip)
         self.audit.write("device.link", login=login, device_id=device_id,
                          device_name=device_name, token_name=token_name,
                          client_ip=client_ip)
-        return 200, {"gitea_base": self.public_gitea, "login": login,
-                     "token": gitea_token, "token_name": token_name,
-                     "device_id": device_id}
+        resp = {"gitea_base": self.public_gitea, "login": login,
+                "token": gitea_token, "token_name": token_name,
+                "device_id": device_id}
+        if granted:
+            resp["granted_repos"] = granted
+        return 200, resp
 
     # -- device registry endpoints -----------------------------------------
 
@@ -862,6 +946,169 @@ class LinkService:
             return None
         return f"forge refused token deletion (HTTP {status}): {resp}"
 
+    # -- sharing -----------------------------------------------------------
+
+    PERMISSIONS = ("read", "write", "admin")
+
+    def _repo_admin_check(self, token, full_name):
+        """Can this caller administer that repo? Ask the FORGE.
+
+        Listing collaborators requires repo admin, so a 200 here is Gitea's own
+        answer to 'may you share this?'. Deciding it ourselves from the repo
+        name would be a second opinion about someone else's authorisation --
+        and the wrong one the first time a repo is transferred.
+        """
+        status, _ = http_json(
+            "GET", f"{self.gitea}/api/v1/repos/{full_name}/collaborators",
+            headers={"Authorization": f"token {token}"})
+        return status == 200
+
+    def grants(self, body, client_ip=None):
+        """Share a repo with someone who already has a forge account."""
+        login = self.whoami(body.get("token"))
+        if not login:
+            return 401, {"error": "invalid or revoked token"}
+        token = body.get("token")
+        action = (body.get("action") or "add").lower()
+        if action not in ("add", "remove", "list"):
+            return 400, {"error": "action must be add, remove or list"}
+        repo = str(body.get("repo") or "").strip()
+        if not repo:
+            return 400, {"error": "repo required"}
+        full = repo if "/" in repo else f"{login}/{repo}"
+        if not self._repo_admin_check(token, full):
+            self.audit.write("grant.denied", login=login, repo=full,
+                             client_ip=client_ip,
+                             reason="caller cannot administer this repo")
+            return 403, {"error": f"you cannot administer {full}"}
+        if action == "list":
+            status, resp = http_json(
+                "GET", f"{self.gitea}/api/v1/repos/{full}/collaborators",
+                headers={"Authorization": f"token {token}"})
+            people = [u.get("login") for u in resp] if isinstance(resp, list) else []
+            return 200, {"repo": full, "collaborators": people}
+
+        who = str(body.get("login") or "").strip()
+        if not who:
+            return 400, {"error": "login required"}
+        permission = (body.get("permission") or "write").lower()
+        if permission not in self.PERMISSIONS:
+            return 400, {"error": f"permission must be one of "
+                                  f"{', '.join(self.PERMISSIONS)}"}
+        if action == "add":
+            status, resp = http_json(
+                "PUT",
+                f"{self.gitea}/api/v1/repos/{full}/collaborators/"
+                f"{urllib.parse.quote(who, safe='')}",
+                headers={"Authorization": f"token {token}"},
+                body={"permission": permission})
+            ok = status in (200, 204)
+        else:
+            status, resp = http_json(
+                "DELETE",
+                f"{self.gitea}/api/v1/repos/{full}/collaborators/"
+                f"{urllib.parse.quote(who, safe='')}",
+                headers={"Authorization": f"token {token}"})
+            ok = status in (200, 204, 404)  # already gone is the wanted state
+        if not ok:
+            self.audit.write("grant.failed", login=login, repo=full,
+                             grantee=who, action=action, client_ip=client_ip,
+                             reason=f"HTTP {status}")
+            return 502, {"error": f"forge refused (HTTP {status}): {resp}"}
+        self.audit.write(f"grant.{action}", login=login, repo=full,
+                         grantee=who, permission=permission,
+                         client_ip=client_ip)
+        return 200, {"repo": full, "login": who, "action": action,
+                     "permission": permission}
+
+    def invite(self, body, client_ip=None):
+        """Promise access to someone who has no forge account yet.
+
+        Nothing is created for them here -- no account, no token. The grant is
+        recorded against their VERIFIED email and applied the first time they
+        link. If they never link, nothing ever existed.
+        """
+        login = self.whoami(body.get("token"))
+        if not login:
+            return 401, {"error": "invalid or revoked token"}
+        email = str(body.get("email") or "").strip().lower()
+        if "@" not in email:
+            return 400, {"error": "a valid email is required"}
+        repos = body.get("repos")
+        if not isinstance(repos, list) or not repos:
+            return 400, {"error": "repos must be a non-empty list"}
+        wanted = []
+        for entry in repos:
+            if isinstance(entry, str):
+                entry = {"name": entry}
+            if not isinstance(entry, dict) or not entry.get("name"):
+                return 400, {"error": "each repo needs a name"}
+            name = str(entry["name"]).strip()
+            full = name if "/" in name else f"{login}/{name}"
+            permission = (entry.get("permission") or "write").lower()
+            if permission not in self.PERMISSIONS:
+                return 400, {"error": f"permission must be one of "
+                                      f"{', '.join(self.PERMISSIONS)}"}
+            if not self._repo_admin_check(body.get("token"), full):
+                self.audit.write("invite.denied", login=login, repo=full,
+                                 invitee=email, client_ip=client_ip,
+                                 reason="caller cannot administer this repo")
+                return 403, {"error": f"you cannot administer {full}"}
+            wanted.append({"repo": full, "permission": permission})
+        with self.state.lock:
+            self.state.add_invite(email, wanted, login)
+        self.audit.write("invite", login=login, invitee=email,
+                         repos=[w["repo"] for w in wanted], client_ip=client_ip)
+        return 200, {"invited": email,
+                     "repos": wanted,
+                     "note": "applied the first time they sign in with a "
+                             "verified email that matches"}
+
+    def apply_invites(self, login, userinfo, client_ip=None):
+        """Turn recorded invites into real collaborator rows at link time.
+
+        Uses the ADMIN credential deliberately: the inviter authorised this
+        when they issued the invite, and their session is long gone by now.
+        Nothing here can fail the link -- someone signing in must not be
+        blocked because a repo they were promised has since been deleted.
+        """
+        email = (userinfo.get("email") or "").strip().lower()
+        if not email or not userinfo.get("email_verified"):
+            # Unverified email must never collect an invite: the address is
+            # the only thing tying the promise to this person.
+            return []
+        with self.state.lock:
+            pending = self.state.peek_invites(email)
+        applied = []
+        # PEEK, not take. Consuming the invite first means a transient 502 from
+        # the forge destroys it: the person links successfully, gets no access,
+        # and re-linking never retries because the promise is gone. Only the
+        # grants that actually landed are removed, so a failure is retried on
+        # the next link instead of being silently lost. (codex review, P2.)
+        for grant in pending:
+            status, resp = http_json(
+                "PUT",
+                f"{self.gitea}/api/v1/repos/{grant['repo']}/collaborators/"
+                f"{urllib.parse.quote(login, safe='')}",
+                headers={"Authorization": _basic(self.cfg["admin_login"],
+                                                 self.cfg["admin_password"])},
+                body={"permission": grant.get("permission", "write")})
+            if status in (200, 204):
+                applied.append(grant["repo"])
+                with self.state.lock:
+                    self.state.consume_invite(email, grant["repo"])
+                self.audit.write("invite.applied", login=login,
+                                 repo=grant["repo"],
+                                 permission=grant.get("permission"),
+                                 client_ip=client_ip)
+            else:
+                self.audit.write("invite.apply.failed", login=login,
+                                 repo=grant["repo"], client_ip=client_ip,
+                                 reason=f"HTTP {status}: {resp}")
+                LOG.warning("invite for %s on %s could not be applied "
+                            "(HTTP %s)", login, grant["repo"], status)
+        return applied
+
     def audit_read(self, body, client_ip=None):
         login = self.whoami(body.get("token"))
         if not login:
@@ -982,6 +1229,10 @@ class Handler(BaseHTTPRequestHandler):
             status, resp = self.service.devices(body, peer)
         elif self.path == "/v1/devices/revoke":
             status, resp = self.service.revoke_device(body, peer)
+        elif self.path == "/v1/grants":
+            status, resp = self.service.grants(body, peer)
+        elif self.path == "/v1/invite":
+            status, resp = self.service.invite(body, peer)
         elif self.path == "/v1/audit":
             status, resp = self.service.audit_read(body, peer)
         else:

tests/test_client.py

@@ -1229,5 +1229,134 @@ class TestWorkspaceBootstrap(GitScenarioBase):
             client.cmd_bootstrap(self.ns())
         self.assertIn("no workspace.json", str(e.exception))
 
+
+class TestResolveGranted(unittest.TestCase):
+    """`get <name>` has to find the repo the forge actually grants.
+
+    Found by running the real flow against production: of 172 repos granted
+    to this estate's own admin, 157 are owned by an ORG, so a bare name failed
+    for 91% of what the user could see -- and the error read "Repository not
+    found", which sounds like a permissions problem rather than a naming one.
+    """
+
+    def setUp(self):
+        self.cfg = {"login": "alice", "gitea_base": "http://forge.example",
+                    "token": "t"}
+        self.repos = [{"full_name": "Nirlabinc/Ai-Assistant"},
+                      {"full_name": "alice/notes"},
+                      {"full_name": "Shreai/notes"}]
+
+    def test_a_bare_name_finds_an_org_owned_repo(self):
+        self.assertEqual(
+            client.resolve_granted(self.cfg, "Ai-Assistant", self.repos),
+            "Nirlabinc/Ai-Assistant")
+
+    def test_an_owner_qualified_name_is_taken_as_given(self):
+        self.assertEqual(
+            client.resolve_granted(self.cfg, "Nirlabinc/Ai-Assistant",
+                                   self.repos),
+            "Nirlabinc/Ai-Assistant")
+
+    def test_an_ambiguous_bare_name_is_refused_with_the_candidates(self):
+        """Two teams with a repo called `notes` is not a guess worth making
+        on someone's behalf."""
+        with self.assertRaises(SystemExit) as e:
+            client.resolve_granted(self.cfg, "notes", self.repos)
+        msg = str(e.exception)
+        self.assertIn("ambiguous", msg)
+        self.assertIn("Shreai/notes", msg)
+        self.assertIn("alice/notes", msg)
+
+    def test_an_unknown_name_falls_back_to_your_own_namespace(self):
+        self.assertEqual(
+            client.resolve_granted(self.cfg, "brand-new", self.repos),
+            "alice/brand-new")
+
+    def test_an_unreachable_listing_does_not_block_a_clone(self):
+        """A network blip must not stop someone cloning a repo they named."""
+        def boom(cfg):
+            raise SystemExit("listing repos failed (HTTP 502)")
+        with mock.patch.object(client, "list_repos", boom), \
+             mock.patch("sys.stdout", new_callable=io.StringIO):
+            self.assertEqual(client.resolve_granted(self.cfg, "notes"),
+                             "alice/notes")
+            self.assertEqual(
+                client.resolve_granted(self.cfg, "Nirlabinc/Ai-Assistant"),
+                "Nirlabinc/Ai-Assistant")
+
+
+class TestDeviceFlowOutputIsVisible(unittest.TestCase):
+    """The code has a five-minute life. If it is sitting in a buffer, the user
+    never sees it and it expires -- which is what happened on 2026-08-23 while
+    driving a first real sign-in through a wrapper."""
+
+    def test_the_url_and_code_are_flushed_immediately(self):
+        seen = []
+        real_print = print
+
+        def spy(*a, **kw):
+            seen.append(kw.get("flush", False))
+            return real_print(*a, **{k: v for k, v in kw.items() if k != "flush"})
+
+        resp = {"verification_uri_complete": "https://id.example/device?user_code=AB-CD",
+                "user_code": "AB-CD", "device_code": "dc", "interval": 0,
+                "expires_in": 0}
+        with mock.patch.object(client, "http_json", lambda *a, **k: (200, resp)), \
+             mock.patch("builtins.print", spy), \
+             mock.patch.object(client.time, "sleep", lambda *_: None):
+            with self.assertRaises(SystemExit):   # expires_in 0 -> times out
+                client.device_flow()
+        self.assertTrue(seen, "device_flow printed nothing")
+        self.assertTrue(all(seen[:2]),
+                        "the URL and code must be printed with flush=True")
+
+
+class TestCodexReviewFindings(GitScenarioBase):
+    """Regressions for the three issues codex found in today's merged work."""
+
+    def test_staged_only_work_survives_a_snapshot(self):
+        """[P2] Stage a hunk, edit further, lose the laptop: the staged version
+        must still be recoverable, not just the later worktree one."""
+        self.write(self.local, "a.txt", "committed")
+        run_git(self.local, "add", "-A")
+        run_git(self.local, "commit", "-m", "base")
+        self.write(self.local, "a.txt", "THE CAREFULLY STAGED VERSION")
+        run_git(self.local, "add", "a.txt")          # staged
+        self.write(self.local, "a.txt", "later scratch edit")   # worktree moved on
+
+        commit, tree = client.build_snapshot(self.local)
+
+        # the worktree state is the snapshot's own tree
+        self.assertEqual(run_git(self.local, "show", f"{commit}:a.txt"),
+                         "later scratch edit")
+        # ...and the staged state is reachable through the extra parent
+        parents = run_git(self.local, "log", "-1", "--format=%P", commit).split()
+        staged = [p for p in parents
+                  if run_git(self.local, "show", f"{p}:a.txt")
+                  == "THE CAREFULLY STAGED VERSION"]
+        self.assertTrue(staged, f"staged version unreachable from {parents}")
+
+    def test_a_snapshot_does_not_disturb_the_index(self):
+        self.write(self.local, "a.txt", "one")
+        run_git(self.local, "add", "-A")
+        run_git(self.local, "commit", "-m", "base")
+        self.write(self.local, "a.txt", "staged")
+        run_git(self.local, "add", "a.txt")
+        before = run_git(self.local, "status", "--porcelain")
+        client.build_snapshot(self.local)
+        self.assertEqual(run_git(self.local, "status", "--porcelain"), before)
+
+    def test_a_truncated_listing_refuses_instead_of_guessing(self):
+        """[P3] A name that is merely beyond the page cap must not resolve to a
+        different repo that happens to exist under your own account."""
+        cfg = {"login": "alice", "gitea_base": "http://forge.example", "token": "t"}
+        with self.assertRaises(SystemExit) as e:
+            client.resolve_granted(cfg, "notes", repos=[], truncated=True)
+        self.assertIn("truncated", str(e.exception))
+        # a complete listing still falls back, because absence is then real
+        self.assertEqual(
+            client.resolve_granted(cfg, "notes", repos=[], truncated=False),
+            "alice/notes")
+
 if __name__ == "__main__":
     unittest.main()

tests/test_server.py

@@ -51,6 +51,14 @@ class StubUpstream(BaseHTTPRequestHandler):
             if not login:
                 return self._json(401, {"message": "unauthorized"})
             return self._json(200, {"login": login})
+        if self.path.endswith("/collaborators") and self.path.startswith("/api/v1/repos/"):
+            full = self.path.split("/api/v1/repos/")[1].rsplit("/collaborators", 1)[0]
+            tok = self.headers.get("Authorization", "").replace("token ", "")
+            caller = st["tokens_by_sha"].get(tok)
+            if st["repo_owner"].get(full) != caller:
+                return self._json(403, {"message": "must have admin rights"})
+            return self._json(200, [{"login": w}
+                                    for w in st["collab"].get(full, {})])
         if self.path.startswith("/api/v1/users/") and not self.path.endswith("/tokens"):
             login = self.path.rsplit("/", 1)[1]
             if login in st["hide_once"]:
@@ -69,6 +77,10 @@ class StubUpstream(BaseHTTPRequestHandler):
         if self.path == "/api/v1/admin/users":
             if body["username"] in st["users"]:
                 return self._json(409, {"message": "user already exists"})
+            if body.get("email") in st["users"].values():
+                # real Gitea: emails are unique across accounts
+                return self._json(422, {"message":
+                    f"e-mail already in use [email: {body['email']}]"})
             st["users"][body["username"]] = body["email"]
             st["created"].append(body)
             return self._json(201, {"login": body["username"]})
@@ -96,8 +108,36 @@ class StubUpstream(BaseHTTPRequestHandler):
                                     f"alice/{body['name']}"})
         self._json(404, {})
 
+    def do_PUT(self):
+        """Gitea adds a collaborator with PUT, and the admin path uses basic
+        auth because the inviter's session is long gone by then."""
+        st = self.state
+        length = int(self.headers.get("Content-Length", 0))
+        body = json.loads(self.rfile.read(length) or b"{}")
+        if "/collaborators/" in self.path and self.path.startswith("/api/v1/repos/"):
+            full = self.path.split("/api/v1/repos/")[1].split("/collaborators/")[0]
+            who = self.path.rsplit("/", 1)[1]
+            if full not in st["repo_owner"]:
+                return self._json(404, {"message": "no such repo"})
+            tok = self.headers.get("Authorization", "")
+            caller = st["tokens_by_sha"].get(tok.replace("token ", ""))
+            if not tok.startswith("Basic ") and st["repo_owner"].get(full) != caller:
+                return self._json(403, {"message": "forbidden"})
+            st["collab"].setdefault(full, {})[who] = body.get("permission", "write")
+            return self._json(204, {})
+        self._json(404, {})
+
     def do_DELETE(self):
         st = self.state
+        if "/collaborators/" in self.path and self.path.startswith("/api/v1/repos/"):
+            full = self.path.split("/api/v1/repos/")[1].split("/collaborators/")[0]
+            who = self.path.rsplit("/", 1)[1]
+            tok = self.headers.get("Authorization", "").replace("token ", "")
+            caller = st["tokens_by_sha"].get(tok)
+            if st["repo_owner"].get(full) != caller:
+                return self._json(403, {"message": "forbidden"})
+            st["collab"].get(full, {}).pop(who, None)
+            return self._json(204, {})
         if self.path.startswith("/api/v1/users/") and "/tokens/" in self.path:
             if not self.headers.get("Authorization", "").startswith("Basic "):
                 # matches real Gitea: token auth cannot delete tokens
@@ -125,7 +165,8 @@ class ServiceTestBase(unittest.TestCase):
         StubUpstream.state = {"users": {}, "created": [], "repos": set(),
                               "token_reqs": [], "hide_once": set(),
                               "tokens": {}, "tokens_by_sha": {},
-                              "revoke_fails": False}
+                              "revoke_fails": False,
+                              "repo_owner": {}, "collab": {}}
         self.upstream = ThreadingHTTPServer(("127.0.0.1", 0), StubUpstream)
         threading.Thread(target=self.upstream.serve_forever, daemon=True).start()
         self.addCleanup(self.upstream.shutdown)
@@ -784,6 +825,214 @@ class TestDeviceEndpointsOverHttp(HandlerTestBase):
         for route, (limit, _window) in granthi_link.DEFAULT_RATE_RULES.items():
             self.assertGreater(limit, 0, f"{route} is disabled by its rule")
 
+
+class TestSharing(ServiceTestBase):
+    """Give a person access to one repo -- and to nothing else."""
+
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+        self.alice = self.stub_link("s1", "alice", email="[email protected]",
+                                    verified=True, device_id="dev-a")[1]
+        self.bob = self.stub_link("s2", "bob", email="[email protected]",
+                                  verified=True, device_id="dev-b")[1]
+        StubUpstream.state["repo_owner"]["alice/notes"] = "alice"
+        StubUpstream.state["repo_owner"]["bob/private"] = "bob"
+
+    def test_owner_can_share_and_unshare(self):
+        status, resp = self.svc.grants({"token": self.alice["token"],
+                                        "repo": "notes", "login": "bob"})
+        self.assertEqual(status, 200, resp)
+        self.assertIn("bob", StubUpstream.state["collab"]["alice/notes"])
+        status, _ = self.svc.grants({"token": self.alice["token"],
+                                     "action": "remove",
+                                     "repo": "notes", "login": "bob"})
+        self.assertEqual(status, 200)
+        self.assertNotIn("bob", StubUpstream.state["collab"]["alice/notes"])
+
+    def test_a_non_owner_cannot_share_someone_elses_repo(self):
+        """The forge decides who may share, not this service."""
+        status, resp = self.svc.grants({"token": self.bob["token"],
+                                        "repo": "alice/notes",
+                                        "login": "bob"})
+        self.assertEqual(status, 403)
+        self.assertEqual(StubUpstream.state["collab"].get("alice/notes", {}), {})
+
+    def test_permission_is_validated(self):
+        status, _ = self.svc.grants({"token": self.alice["token"],
+                                     "repo": "notes", "login": "bob",
+                                     "permission": "owner"})
+        self.assertEqual(status, 400)
+
+    def test_listing_shows_who_has_access(self):
+        self.svc.grants({"token": self.alice["token"], "repo": "notes",
+                         "login": "bob"})
+        status, resp = self.svc.grants({"token": self.alice["token"],
+                                        "action": "list", "repo": "notes"})
+        self.assertEqual((status, resp["collaborators"]), (200, ["bob"]))
+
+    def test_removing_someone_who_never_had_access_is_not_an_error(self):
+        status, _ = self.svc.grants({"token": self.alice["token"],
+                                     "action": "remove", "repo": "notes",
+                                     "login": "nobody"})
+        self.assertEqual(status, 200)
+
+
+class TestInvites(ServiceTestBase):
+    """A promise of access for someone with no account yet."""
+
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+        self.alice = self.stub_link("s1", "alice", email="[email protected]",
+                                    verified=True, device_id="dev-a")[1]
+        StubUpstream.state["repo_owner"]["alice/notes"] = "alice"
+
+    def invite(self, email, repos=("notes",), token=None):
+        return self.svc.invite({"token": token or self.alice["token"],
+                                "email": email, "repos": list(repos)})
+
+    def test_invite_creates_nothing_until_they_sign_in(self):
+        status, resp = self.invite("[email protected]")
+        self.assertEqual(status, 200, resp)
+        # no account, no collaborator row yet
+        self.assertNotIn("carol", StubUpstream.state["users"])
+        self.assertEqual(StubUpstream.state["collab"].get("alice/notes", {}), {})
+
+    def test_access_lands_on_first_link_with_a_verified_email(self):
+        self.invite("[email protected]")
+        status, resp = self.stub_link("s9", "carol", email="[email protected]",
+                                      verified=True, device_id="dev-c")
+        self.assertEqual(status, 200, resp)
+        self.assertEqual(resp.get("granted_repos"), ["alice/notes"])
+        self.assertIn("carol", StubUpstream.state["collab"]["alice/notes"])
+
+    def test_an_unverified_email_collects_nothing(self):
+        """The address is the only thing tying the promise to this person."""
+        self.invite("[email protected]")
+        status, resp = self.stub_link("s10", "dave", email="[email protected]",
+                                      verified=False, device_id="dev-d")
+        self.assertEqual(status, 200)
+        self.assertIsNone(resp.get("granted_repos"))
+        self.assertEqual(StubUpstream.state["collab"].get("alice/notes", {}), {})
+        # and the invite is still waiting, not consumed
+        self.assertTrue(self.svc.state.peek_invites("[email protected]"))
+
+    def test_an_invite_is_applied_once(self):
+        self.invite("[email protected]")
+        self.stub_link("s11", "erin", email="[email protected]", verified=True,
+                       device_id="dev-e")
+        status, resp = self.stub_link("s11", "erin", email="[email protected]",
+                                      verified=True, device_id="dev-e2")
+        self.assertIsNone(resp.get("granted_repos"))
+
+    def test_cannot_invite_to_a_repo_you_do_not_administer(self):
+        StubUpstream.state["repo_owner"]["bob/secret"] = "bob"
+        status, _ = self.invite("[email protected]", repos=("bob/secret",))
+        self.assertEqual(status, 403)
+        self.assertEqual(self.svc.state.peek_invites("[email protected]"), [])
+
+    def test_reinviting_updates_the_permission_instead_of_stacking(self):
+        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
+                         "repos": [{"name": "notes", "permission": "read"}]})
+        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
+                         "repos": [{"name": "notes", "permission": "write"}]})
+        pending = self.svc.state.peek_invites("[email protected]")
+        self.assertEqual(len(pending), 1)
+        self.assertEqual(pending[0]["permission"], "write")
+
+    def test_a_bad_email_is_refused(self):
+        status, _ = self.svc.invite({"token": self.alice["token"],
+                                     "email": "not-an-email",
+                                     "repos": ["notes"]})
+        self.assertEqual(status, 400)
+
+    def test_a_failed_grant_does_not_block_the_person_signing_in(self):
+        """Someone must not be locked out because a repo they were promised
+        has since been deleted."""
+        self.invite("[email protected]", repos=("notes",))
+        StubUpstream.state["repo_owner"].pop("alice/notes")
+        status, resp = self.stub_link("s12", "gina", email="[email protected]",
+                                      verified=True, device_id="dev-g")
+        self.assertEqual(status, 200)
+        self.assertIn("token", resp)
+
+    def test_invite_and_application_are_both_audited(self):
+        self.invite("[email protected]")
+        self.stub_link("s13", "hana", email="[email protected]", verified=True,
+                       device_id="dev-h")
+        _, mine = self.svc.audit_read({"token": self.alice["token"]})
+        self.assertIn("invite", [e["event"] for e in mine["events"]])
+
+
+class TestInviteSurvivesAFailedGrant(ServiceTestBase):
+    """[P2, codex] A transient forge error must not destroy the promise."""
+
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+        self.alice = self.stub_link("s1", "alice", email="[email protected]",
+                                    verified=True, device_id="dev-a")[1]
+        StubUpstream.state["repo_owner"]["alice/notes"] = "alice"
+        StubUpstream.state["repo_owner"]["alice/reports"] = "alice"
+
+    def test_a_failed_grant_leaves_the_invite_pending_for_next_time(self):
+        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
+                         "repos": ["notes"]})
+        # the forge loses the repo mid-flight -> the PUT 404s
+        StubUpstream.state["repo_owner"].pop("alice/notes")
+        status, resp = self.stub_link("s9", "carol", email="[email protected]",
+                                      verified=True, device_id="dev-c")
+        self.assertEqual(status, 200)                 # sign-in still succeeds
+        self.assertIsNone(resp.get("granted_repos"))
+        # the promise is STILL THERE rather than silently consumed
+        self.assertEqual([g["repo"] for g in self.svc.state.peek_invites("[email protected]")],
+                         ["alice/notes"])
+        # and it lands on the next link, once the repo is back
+        StubUpstream.state["repo_owner"]["alice/notes"] = "alice"
+        status, resp = self.stub_link("s9", "carol", email="[email protected]",
+                                      verified=True, device_id="dev-c2")
+        self.assertEqual(resp.get("granted_repos"), ["alice/notes"])
+        self.assertEqual(self.svc.state.peek_invites("[email protected]"), [])
+
+    def test_a_partial_failure_only_consumes_what_landed(self):
+        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
+                         "repos": ["notes", "reports"]})
+        StubUpstream.state["repo_owner"].pop("alice/reports")   # one of two fails
+        _, resp = self.stub_link("s10", "dan", email="[email protected]",
+                                 verified=True, device_id="dev-d")
+        self.assertEqual(resp.get("granted_repos"), ["alice/notes"])
+        self.assertEqual([g["repo"] for g in self.svc.state.peek_invites("[email protected]")],
+                         ["alice/reports"])
+
+
+class TestDuplicateEmailIsExplained(ServiceTestBase):
+    """A 502 sends someone looking for an outage. The real answer is that the
+    address already belongs to another account here -- say so. (Hit live on
+    2026-08-23 when an operator moved an email onto a different account.)"""
+
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+        # an existing account already holds the address
+        StubUpstream.state["users"]["existing"] = "[email protected]"
+
+    def test_it_is_a_409_that_names_the_problem(self):
+        status, resp = self.stub_link("s-new", "brandnew", email="[email protected]",
+                                      verified=True, device_id="dev-x")
+        self.assertEqual(status, 409, resp)
+        msg = resp["error"]
+        self.assertIn("[email protected]", msg)
+        self.assertIn("already belongs to a different account", msg)
+        self.assertIn("brandnew", msg)          # names the login it tried
+        self.assertNotIn("502", msg)
+
+    def test_a_normal_create_is_unaffected(self):
+        status, resp = self.stub_link("s-ok", "fresh", email="[email protected]",
+                                      verified=True, device_id="dev-y")
+        self.assertEqual(status, 200, resp)
+        self.assertIn("token", resp)
+
 if __name__ == "__main__":
     unittest.main()