test: preserve empty credential helper reset

Merge pull request 'fix(link): say "that address is already an account here", not "502"' (#10 ) from fix/link-duplicate-email-message into main
fix(link): say "that address is already an account here", not "502"
2026-08-24 03:30:14 -04:00 · 2026-08-23 16:46:22 -04:00 · 2026-08-23 16:44:49 -04:00 · 2026-08-23 16:14:56 -04:00 · 2026-08-23 16:14:21 -04:00 · 2026-08-23 15:58:36 -04:00
7 changed files with 4440 additions and 95 deletions
@@ -1,4 +1,4 @@
-# granthi-sync v1
+# 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 +
@@ -29,8 +29,10 @@ git CLI only — same portability heritage as the estate's `gitea_sync.py` mesh.
 ## Quickstart (invited user)

 You need a shre-id account — an operator creates it; there is no open signup
-(see "Invite-only story"). You also need to be on the tailnet: the
-provisioning service is not publicly exposed yet.
+(see "Invite-only story"). You do **not** need to be on any private network:
+the provisioning service answers at `https://granthi-link.shre.ai`, which is
+the client's default server. Internal machines can still pass
+`--server http://100.111.127.127:3042` to reach it over the tailnet.

 ```sh
 git clone https://granthi.shre.ai/nirpa/granthi-sync.git
@@ -50,10 +52,17 @@ cd granthi-sync
 # 3b. ...or push a local folder up. It becomes a private repo.
 ./bin/granthi-sync add ~/work/notes

+# 3c. ...or pull down everything this account is allowed to see.
+./bin/granthi-sync get --all --into ~/granthi
+
 # 4. Keep everything synced. Autocommits, ff-pulls, pushes; skips anything
 #    that has diverged rather than merging or forcing.
 ./bin/granthi-sync watch                # --once for a single pass
-./bin/granthi-sync status               # what is linked, last sync, divergence
+./bin/granthi-sync status               # what is linked, mode, last sync
+
+# 5. Go back to how a folder looked at some point in time.
+./bin/granthi-sync snapshots ~/work/notes
+./bin/granthi-sync restore  ~/work/notes --at 20260823T142530Z
 ```

 Run `watch` as a background daemon on macOS with
@@ -67,6 +76,111 @@ no account, no token, no partial state. Just run `link` again.
 to merge; when a folder shows `DIVERGED` in `status`, resolve it in git or on
 the forge web UI. The client will never force or auto-merge your work.

+## Two modes, because two very different folders ask for this
+
+A folder people sync is either *their documents* or *their git project*, and
+the correct behaviour is opposite in each case. Each linked folder therefore
+carries a `mode`.
+
+| | `mirror` | `snapshot` |
+|---|---|---|
+| chosen for | a plain folder `add` turned into a repo | a folder that was already a git repo, and anything `get` clones |
+| commits on your behalf | yes, `sync: <ISO ts>` | **never** |
+| where work lands | the branch | `refs/granthi-backup/<device>/<ts>` |
+| a restore point is | every commit | every snapshot |
+
+`snapshot` mode is what "the work may not be committed, but it is still
+backed up" means in git terms. Each pass loads a scratch index from HEAD,
+stages the working tree into *that* index, writes a tree, and commits it with
+`commit-tree`. HEAD, your index, your stash and every file on disk are
+untouched — you can be mid-rebase with a dirty tree and the backup still
+records exactly what is on the disk right now. The user's history stays the
+user's.
+
+Why a custom ref namespace: verified on the beta forge (Gitea 1.27.2) that
+`refs/granthi-backup/...` is accepted, is readable through `ls-remote`, and
+does **not** appear in the branch list. Under `refs/heads` a machine taking a
+backup every 30 seconds would bury the branches a person actually made.
+
+Snapshots are parented on HEAD and deliberately **not** chained to the
+previous snapshot: chaining would keep every old snapshot reachable from the
+newest, so pruning a ref would free nothing and retention would be
+decorative.
+
+**Retention** (or 30-second backups become a disk leak nobody can navigate):
+everything is kept for 24 h, then thinned to hourly for 7 days, then daily.
+Pruning runs at most hourly, per device, and only over that device's own
+refs. A ref whose timestamp this version cannot parse is **kept** — deleting
+the unrecognised is how a backup system loses the one thing someone needed.
+
+**Restore never writes over the working tree.** `restore` materialises a
+restore point into a *new* directory and refuses a non-empty destination.
+Someone restoring a backup is already having a bad day; overwriting the files
+they still have would make the recovery tool the second disaster.
+
+## The credential helper must be the ONLY helper (found by live QA)
+
+`credential.helper` is a list that accumulates across system, global and repo
+config, and git asks every helper in it. A stock mac already has two —
+`osxkeychain` from Xcode's gitconfig, and `store` from many people's
+`~/.gitconfig` — and they lose in both directions:
+
+* **reading:** a stale entry for the forge host answers before our helper, so
+  pushes fail `remote: Failed to authenticate user` long after the token was
+  rotated, and nothing in this tool's config explains why. This is exactly
+  how the first live-QA run failed;
+* **writing:** git calls `approve` on every helper after a successful auth,
+  so `store` copies the forge token into `~/.git-credentials` **in
+  plaintext**. Keeping the token in a 0600 file and out of remote URLs buys
+  nothing if git then hands it to a plaintext store.
+
+So `install_credential_helper` (and the `git clone` in `get`) sets an **empty**
+`credential.helper` first, which resets the inherited list, then adds ours.
+Exactly one helper serves this repo.
+
+Corollary worth remembering: a token embedded in a remote URL gets saved by
+`store` on first use. During QA a verification clone with a URL-embedded
+token re-created the very entry that had just been cleaned out. That is the
+whole reason this client passes tokens through a helper and never a URL.
+
+## Device identity
+
+`link` mints a uuid on first run and persists it in `~/.granthi-sync/config.json`
+as `device_id`, and sends it to `/v1/link`. Hostnames are neither stable
+(people rename laptops) nor unique (every new mac is "Mac mini"), so a
+hostname cannot key a backup ref or a device registry — two machines would
+overwrite each other's snapshots. The service-side device registry is the
+next phase; the client leads so the id already exists when it lands.
+
+## What "add the computer to the network" means — and does not
+
+The onboarding shape is: download → login → **the device is federated to the
+account** → the device can reach its repos.
+
+The middle step is a *device registration*, not a network membership. Those
+sound like one step and must not be built as one: this estate's tailnet is a
+single flat private network carrying the granthi VPS, aros-vps, the Shadow
+box and the Mac. Putting a customer's laptop on it to let them sync a folder
+would hand that laptop L3 reach to every piece of infrastructure we run.
+
+So:
+
+* **Our own machines** may join the tailnet — that is an operator action with
+  an operator's judgement behind it.
+* **Customer devices never do.** Their transport is public HTTPS to
+  granthi-link and the forge through cloudflared. **Done 2026-08-23:**
+  `https://granthi-link.shre.ai` fronts `:3042` on the `pulse-granthi-edge`
+  tunnel, so a new computer signs itself in with one command and never
+  touches the private network.
+
+  Two things that bit during that rollout, worth not rediscovering:
+  `link.granthi.shre.ai` fails TLS — Cloudflare's Universal SSL covers
+  `shre.ai` and `*.shre.ai`, **not** a third-level `*.granthi.shre.ai`, so
+  the hostname has to be second-level. And exposure REQUIRES
+  `trust_forwarded_for: true` with `trusted_proxies: ["100.107.37.98/32"]`
+  in the same change: behind the tunnel every request otherwise looks like
+  the tunnel itself, and one abuser would spend everybody's rate budget.
+
 ## Components

 ### `server/granthi_link.py` — provisioning service (granthi VPS)
@@ -90,6 +204,59 @@ window. 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.

+#### Rate limiting
+
+Sliding-window, per `(route, client)`, **on by default** — `/v1/link` creates
+accounts and mints tokens, so unlimited has to be a deliberate config act,
+never an omission. Defaults: `/v1/link` 5/hour, `/v1/repos` 60/hour. `GET
+/health` is never limited. Over the limit → **429** with a `Retry-After`
+header, decided *before* the body is read so an abusive caller costs nothing.
+
+```json
+"rate_limit": {
+  "enabled": true,
+  "trust_forwarded_for": false,
+  "trusted_proxies": [],
+  "rules": {"/v1/link": [5, 3600], "/v1/repos": [60, 3600]}
+}
+```
+
+* Anything malformed **refuses startup** rather than silently meaning
+  unlimited: a bad rule, a non-boolean `enabled`/`trust_forwarded_for` (JSON
+  `null`, `0`, or the *string* `"false"` — every non-empty string is truthy),
+  or a `rate_limit` that is not an object. `[0, N]` disables an endpoint
+  outright.
+* Denied requests are *not* recorded, so a client that keeps hammering cannot
+  push its own window forward and lock itself out permanently.
+* State is an in-process dict behind a lock — granthi-link is one
+  `ThreadingHTTPServer`, so that is the entire store. **If this ever runs
+  multi-process or multi-host, the limiter must move with it.** The clock is
+  read *inside* the lock; taken outside, racing threads append out of order
+  and both `Retry-After` and window reclamation silently go wrong.
+* The key store is capped (`MAX_RATE_KEYS`) **per route**, not globally. At
+  capacity it reclaims expired windows, and if every window is still live it
+  **refuses the new key** — fail closed. Evicting a live window would let an
+  attacker who can mint many distinct keys clear their *own* limit on demand.
+  The per-route budget matters just as much: with one shared table, a flood
+  of cheap `/v1/repos` keys would exhaust it and lock brand-new `/v1/link`
+  clients out, turning the fail-closed guard into a cross-route DoS.
+* Rule values must be real integers. `bool` subclasses `int` in Python, so
+  `[5, true]` would otherwise pass as a **1-second** window — an hourly limit
+  quietly becoming ~5/sec.
+* `trust_forwarded_for` is **off** by default, and turning it on **requires a
+  non-empty `trusted_proxies`** — the header is honored only when the socket
+  peer is in that list. Without it, anyone reaching the origin directly (it
+  also listens on the tailnet) could pick and rotate their own rate-limit key
+  just by sending a header. Turn it on when exposing behind cloudflared,
+  where every request otherwise arrives from the tunnel and one abuser would
+  starve everyone. A caller can *prepend* anything to `X-Forwarded-For`; a
+  trusted proxy *appends* the peer it actually saw, so the service reads the
+  **last** entry, never the first, and requires it to parse as a real IP.
+  `trusted_proxies` is validated at startup: it must be a list (a bare string
+  would be iterated character by character), every entry a valid network, and
+  wildcards (`0.0.0.0/0`, `::/0`) are refused outright — they would restore
+  exactly the "trust anyone" hole the setting exists to close.
+
 #### Identity binding (`state.json`)

 `/v1/link` originally bound purely by `preferred_username` / email
@@ -158,13 +325,32 @@ deleted it again, `DELETE …/tokens/{id}` returning 204 under basic auth):
  (`authorization_pending`/`slow_down` handled), then calls `/v1/link`.
  `--token` skips the device flow with a ready Zitadel token (headless/dev).
  Result stored in `~/.granthi-sync/config.json` (0600).
-* `list` — every repo the linked token can see, with the local folder each
-  is already synced to. Reads `GET /api/v1/user/repos` on the forge
+* `list [pattern]` — every repo the linked token can see, with the local
+  folder each is already synced to. `pattern` narrows the table by name
+  (substring, or a glob like `work-*`), case-insensitively, against both
+  `owner/name` and the bare name. Filtering is display-only: the set already
+  came from the forge under this account's token. Reads
+  `GET /api/v1/user/repos` on the forge
  **directly** with the scoped user token — no granthi-link round-trip, so
  the read path needs no service change. Pagination is followed to a short
  page; if the `FORGE_MAX_PAGES` guard trips, the output says the list is
  incomplete rather than letting a bounded page read as the whole set.
-* `get <repo|owner/repo> [--into DIR]` — the download half of `add`. Clones
+* `get --all [--into DIR] [--mode M]` — clone every repo this account can
+  see, skipping the ones already linked here. **"Only the repos they are
+  granted" needs no client-side permission logic**: `/api/v1/user/repos` is
+  evaluated by the forge against this account's own scoped token, so the
+  list *is* the grant. A client-side filter would be a second opinion about
+  someone else's authorisation. One repo failing does not abandon the rest,
+  and a truncated listing is reported loudly — `--all` must never quietly
+  mean "the first 2000". `alice/notes` and `bob/notes` both want
+  `<base>/notes`; the second is cloned to `<base>/bob-notes` and the clash is
+  logged, because reporting it as "already present" would leave the user
+  believing they had pulled both.
+* `get <repo|owner/repo> [--into DIR] [--mode M]` — the download half of
+  `add`. Defaults to `snapshot` mode unless the repo carries a
+  `.granthi-sync.json` marker saying otherwise, so a plain synced folder
+  behaves the same on the second machine while someone's real project is
+  never autocommitted onto. Clones
  with `--origin granthi` (the remote name `watch` looks for) and
  `-c credential.helper=…` (the repo does not exist yet, so the helper
  cannot be installed first; git also persists it into the new config), then
@@ -172,22 +358,76 @@ deleted it again, `DELETE …/tokens/{id}` returning 204 under basic auth):
  so a cloned repo is picked up by `watch` immediately. Refuses a non-empty
  destination. Branch is read with `symbolic-ref` (an empty repo has an
  unborn HEAD) and falls back to `main`.
-* `add <folder> [--name N] [--private|--public]` — `git init -b main` if
-  needed, creates the cloud repo via `/v1/repos`, adds remote `granthi`,
-  initial commit + push. The token is delivered by a **git credential
-  helper** (the client's hidden `git-credential` subcommand reading the 0600
-  config) — never embedded in the remote URL (estate rule).
-* `watch [--interval 30] [--once]` — per folder: autocommit
-  (`sync: <ISO ts>`) → fetch → ff-pull if remote strictly ahead → push if
-  local strictly ahead. **DIVERGED → log + record + SKIP. Never force, never
-  merge** — the same policy as the mesh. SIGTERM-clean.
-* `status` — table of linked folders, last sync, divergence flags.
+* `add <folder> [--name N] [--private|--public] [--mode M] [--force]` —
+  `git init -b main` if needed, creates the cloud repo via `/v1/repos`, adds
+  remote `granthi`, initial commit + push. The token is delivered by a **git
+  credential helper** (the client's hidden `git-credential` subcommand
+  reading the 0600 config) — never embedded in the remote URL (estate rule).
+  Pushes the folder's **current** branch, not a hardcoded `main`: an existing
+  repo may sit on `master` or a feature branch, and publishing that work
+  under the wrong name is not a cosmetic error.
+  Two guards, because `add -A` takes whatever it is given: a starter
+  `.gitignore` is seeded when the folder has none (an existing one is never
+  touched — it is the user's), and a folder over 20 000 files / 512 MB is
+  refused unless `--force`. The seeded ignore file covers `.env`, `*.key`,
+  `*.pem`, `id_rsa` and friends, and it governs snapshots too — the scratch
+  index honours `.gitignore` exactly as a normal commit does.
+  The size guard measures what git *would* sync, ignore rules included
+  (including the machine's global excludes), because its own advice is "add
+  a .gitignore for what should not sync" and advice that changes nothing is
+  worse than none. It asks git through a **throwaway git dir outside the
+  folder**, so a refused `add` leaves no `.git` behind in a directory the
+  user never agreed to turn into a repo.
+* `watch [--interval 30] [--once]` — per folder, by mode. `mirror`:
+  autocommit (`sync: <ISO ts>`) → fetch → ff-pull if remote strictly ahead →
+  push if local strictly ahead. `snapshot`: fetch → push a snapshot of the
+  working tree to this device's backup ref → ff-pull only when the tree is
+  clean (local edits are already safe on the backup ref, so it reports and
+  leaves the tree alone rather than failing) → push the user's own commits
+  when they are strictly ahead. **DIVERGED → log + record + SKIP. Never
+  force, never merge** — the same policy as the mesh — **but the backup
+  still happens**, because divergence is when work is most at risk.
+  Retention pruning runs at most hourly. SIGTERM-clean.
+* `snapshots <folder> [--limit 20]` — restore points, newest first, **across
+  every device**, with the device that took each one. Read from the
+  **remote**, not a local cache: the feature exists for the case where this
+  machine is gone.
+
+  The three scopes differ deliberately. Writing is device-scoped, so two
+  machines never overwrite each other. Pruning is device-scoped, so machine A
+  never applies its clock to machine B's refs. **Reading is not scoped** — a
+  replacement laptop has a new id, and scoping the read to it would print
+  "no restore points yet" while the backups sit on the forge. That defect
+  was live in the first draft and is now pinned by a test that restores a
+  dead machine's work from a fresh clone.
+* `restore <folder> --at <ts|sha> [--into DIR]` — materialise one restore
+  point into a new directory; refuses a non-empty destination. Accepts what
+  `snapshots` printed in either mode, including a mirror-mode `%cI`
+  timestamp. Two commits inside the same second share that timestamp, so an
+  ambiguous `--at` is **refused with the candidate ids** rather than
+  resolved by guessing.
+* `status` — table of linked folders, mode, last sync, divergence flags.
 * Run as a daemon on macOS with `client/launchd/ai.granthi.sync.plist`
  (edit the script path, then `launchctl bootstrap gui/$UID <plist>`).

 ## Tests

-* `python3 -m unittest discover -s tests` — 65 tests: autocommit/ff/diverged
+* `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
+  up; a dirty tree blocking the ff-pull but not the backup; retention keeping
+  everything recent, thinning to hourly then daily, and **keeping**
+  unparseable timestamps; prune deleting only the thinned refs; `.gitignore`
+  seeding never overwriting an existing one and keeping `.env` out of
+  snapshots; the folder-size guard being bounded rather than walking the
+  disk; mode detection; `list` filtering; `get --all` skipping what is
+  already present, defaulting to snapshot mode, and shouting about
+  truncation; `restore` writing a new folder, refusing a non-empty
+  destination, and leaving the working tree alone; and the credential helper
+  being the only one the repo consults, proven by driving
+  `git credential fill` against a deliberately poisoned outer helper.
+* Earlier suite: autocommit/ff/diverged
  logic against real temp git repos (including "diverged never touches the
  remote"), config 0600 handling (including umask-proof creation and a
  no-chmod guard), credential-helper quoting/injection, mocked device-flow
@@ -214,10 +454,172 @@ the provisioning service creates their forge account + scoped token on the
 fly — the forge never sees a password and the user never sees the forge admin.
 Every linked folder becomes a private repo under their account.

+## Devices, sign-out, and the audit trail
+
+**`granthi-sync devices`** lists every computer signed in to the account —
+name, id, when it linked, and whether it is still active. The registry hangs
+off the identity that owns it in `state.json`, so "which computers can reach
+my files" has one answer that cannot drift from the identity map.
+
+**`granthi-sync logout [--device ID]`** signs a computer out. Revocation
+happens **at the forge**: granthi-link deletes that device's Gitea token
+(admin basic auth + `Sudo`, the only mechanism Gitea 1.27 accepts — verified
+204, after which the token returns 401 immediately). It is not a flag a
+client could ignore, which is the only kind of sign-out worth having for a
+laptop somebody lost. Signing out the current computer also deletes the local
+token; linked folders are left on disk untouched.
+
+Two properties worth keeping:
+
+* The device id is part of the **token name**, because revocation deletes by
+  name. Without it, one user linking two machines called "macbook" in the
+  same second would collide, and signing one out would kill the other.
+* A failed forge deletion is **not** recorded as revoked. A registry that
+  says "revoked" while the token still works is worse than an honest error.
+
+**`granthi-sync activity`** shows the security events for the account:
+`device.link`, `device.revoke`, `device.revoke.denied`, with timestamp,
+device, and client IP. The log is append-only JSONL (0600, rotated at 64 MB),
+kept **separate** from `state.json` on purpose — state is rewritten
+atomically on every change, and an audit trail the audited thing can rewrite
+is not an audit trail. A failed audit write is logged loudly and never breaks
+the request it was auditing.
+
+**What the audit log cannot see, and this matters.** Every event in it is one
+granthi-link handled. **Git pushes and pulls do not pass through this
+service** — they go straight to the forge — so:
+
+| you want to know | where it actually lives |
+|---|---|
+| who linked/revoked a computer, from what IP | `granthi-sync activity` (this log) |
+| who pushed what, and when | Gitea: the repo's activity feed and commit history |
+| who *pulled* or cloned | **nowhere by default** — Gitea does not record fetches unless its router access log is enabled |
+| last time a device used its token | Gitea `access_token.updated_unix` |
+
+Reading the audit log and believing it lists file activity would be a real
+mistake, so `/v1/audit` returns that caveat in its own response.
+
+### Endpoints added
+
+* `POST /v1/devices {token}` → the caller's devices.
+* `POST /v1/devices/revoke {token, device_id}` → kills that device's forge
+  token. **Never rate-limited** — nobody should be throttled out of signing
+  a lost laptop out.
+* `POST /v1/audit {token, limit}` → the caller's own events.
+
+Authorisation on all three is the same: the caller proves who they are by
+holding a working forge token, and **the forge decides** whose it is
+(`GET /api/v1/user`). No login is ever read from the request body, so a body
+claiming another account changes nothing.
+
+## Workspace bootstrap — what a new computer should pull first
+
+    granthi-sync bootstrap ~/granthi/acme --into ~/work
+
+A workspace is a repo holding a `workspace.json`:
+
+```json
+{"name": "acme",
+ "repos": ["notes", {"name": "reports", "mode": "mirror"}],
+ "apps": [{"id": "dash", "repo": "nirpa/hermes-agent", "model": "aum/70b",
+           "setup": "docs/SETUP.md", "needs_keys": ["anthropic"]}]}
+```
+
+`bootstrap` pulls the repos it names — **still only the ones the forge grants
+this account**; a manifest asking for a repo you cannot see prints
+`NOT GRANTED` and moves on, because that is a permissions answer and not an
+error to route around.
+
+**It does not install applications, and that is the design, not a shortcut.**
+Dash, deck, genie and shiva each have their own repo, deploy path and
+reviewers. A sync client that installed them would create a second,
+unreviewed deploy path beside the real one — the same copy-instead-of-refer
+mistake the estate rulebook exists to end. So `apps` is *referenced*: the
+command prints each app's repo, the model it uses, its setup doc, and which
+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`.
+
+## 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
+inviting somebody, and that person's device waking up with access to *some*
+repos and not others.
+
+Shape this should take, so the next session does not re-litigate it:
+
+* **Where grants live: granthi-link's own store, not Zitadel orgs.** The
+  estate's house pattern is app-side tenancy tables with the IdP only
+  providing identity (see the AROS `tenants` / `tenant_members` split).
+  Grants therefore sit beside `state.json`, and **Gitea is the enforcement
+  point** — a grant is materialised as a repo collaborator or an org team
+  membership, so the forge itself refuses unauthorised reads. Nothing in the
+  client decides access, which is why `get --all` needs no permission logic.
+* **Token scope does not change.** `write:repository,write:user` stays; per
+  repo permission is collaborator/team state, not a token property.
+* **`POST /v1/invite`** (account admin → new member): creates the shre-id
+  user (Zitadel admin PAT, on aros-vps at
+  `/opt/shre-id/deploy/secrets/shre_id_zitadel_pat`), records the intended
+  grants, and returns an invite the person redeems by running `link`. Until
+  they redeem it, nothing exists on the forge.
+* **`POST /v1/grants`** (account admin): add/remove repo access for a member;
+  applies the change to Gitea and records it. Removing a grant must also
+  remove the collaborator — a grant store that drifts from the forge is
+  worse than no store.
+* Both endpoints are account-admin-only and rate-limited like `/v1/link`.
+* The grant store inherits the same fragility already noted for the rate
+  limiter: a flat JSON file behind an in-process lock, fine for one
+  `ThreadingHTTPServer` and **not** fine the day this runs multi-process.
+
 ## Promotion window (beta → prod)

-1. **Expose :3042** behind cloudflared (granthi.shre.ai vhost or
-   link.granthi.shre.ai) — today it is tailnet-only by design.
+1. ~~**Expose :3042** behind cloudflared~~ **DONE 2026-08-23** —
+   `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
@@ -226,8 +628,11 @@ Every linked folder becomes a private repo under their account.
   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 —
   promotion is an operator action, not an agent action).
-4. Add rate limiting / abuse controls before public exposure (one token mint
-   per link call today).
+4. ~~Add rate limiting / abuse controls before public exposure.~~ **DONE** —
+   see "Rate limiting" above (5 links/hour per client by default). When
+   exposing behind cloudflared, set `trust_forwarded_for: true` in the same
+   change, or every request will look like the tunnel and one abuser will
+   throttle everybody.
 5. **Hardening checklist (must all hold before exposing):**
   - [ ] `config.json` is 0600 (or 0400) and owned by the service user —
         the service refuses to start otherwise; verify with
@@ -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.
@@ -6,5 +6,10 @@
  "admin_login": "nirpa",
  "admin_password": "FROM /opt/gitea-beta/.admin-creds (required: Gitea 1.27 token minting only works via basic auth + Sudo header)",
  "binds": [["127.0.0.1", 3042], ["100.111.127.127", 3042]],
-  "test_mode": false
+  "test_mode": false,
+  "rate_limit": {
+    "enabled": true,
+    "trust_forwarded_for": false,
+    "rules": {"/v1/link": [5, 3600], "/v1/repos": [60, 3600]}
+  }
 }
@@ -52,6 +52,7 @@ Stdlib only. Python 3.9+.
 """

 import base64
+import ipaddress
 import json
 import logging
 import os
@@ -61,7 +62,9 @@ import signal
 import string
 import sys
 import threading
+import time
 import urllib.error
+import urllib.parse
 import urllib.request
 from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -71,14 +74,166 @@ LOG = logging.getLogger("granthi-link")

 DEFAULT_CONFIG = "/opt/granthi-link/config.json"
 DEFAULT_STATE = "/opt/granthi-link/state.json"
+DEFAULT_AUDIT = "/opt/granthi-link/audit.jsonl"
 MAX_BODY_BYTES = 64 * 1024
 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._-]+")

+# Rate limiting. /v1/link is the expensive endpoint -- it round-trips Zitadel,
+# can CREATE a forge account and always mints a token -- so its default is
+# deliberately tight. /v1/repos only spends the caller's own token.
+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
+                      # RateLimiter.check), which would mean nobody could
+                      # ever sign a lost laptop out. Caught in live QA --
+                      # unit tests called the service method directly and so
+                      # never went through the limiter at all.
+                      "/v1/devices/revoke": (600, 3600)}
+MAX_RATE_KEYS = 10000
+
+
+# --------------------------------------------------------------------------
+# Rate limiting
+# --------------------------------------------------------------------------
+
+class RateLimiter:
+    """Sliding-window limiter keyed by (route, client).
+
+    In-process and lock-guarded: granthi-link is ONE ThreadingHTTPServer
+    process, so a dict is the entire store -- no redis, no shared cache, and
+    nothing to keep consistent across nodes. If this ever runs multi-process
+    the limiter must move with it; that is why the store is behind this class
+    rather than sprinkled through the handler.
+    """
+
+    def __init__(self, rules=None, max_keys=MAX_RATE_KEYS, clock=None):
+        self.rules = dict(rules or DEFAULT_RATE_RULES)
+        # Per-ROUTE budget, not one global table. A shared cap lets a flood of
+        # cheap /v1/repos keys exhaust the table and lock brand-new /v1/link
+        # clients out -- turning the fail-closed capacity guard into a
+        # cross-route denial of service. Each route gets its own space.
+        self.max_keys = max_keys
+        self._clock = clock or time.monotonic
+        self._hits = {route: {} for route in self.rules}
+        self._lock = threading.Lock()
+
+    def check(self, route, client):
+        """-> (allowed: bool, retry_after: int). Records the hit when allowed.
+
+        Denied requests are NOT recorded: a client that keeps hammering must
+        not push its own window forward and lock itself out indefinitely.
+        """
+        rule = self.rules.get(route)
+        if not rule:
+            return True, 0
+        limit, window = rule
+        if limit <= 0:                      # 0 = endpoint disabled entirely
+            return False, window
+        with self._lock:
+            # Clock read INSIDE the lock: taken outside, two racing threads
+            # can append out of order, and both hits[0] (retry_after) and
+            # v[-1] (reclamation age) assume the list is chronological.
+            now = self._clock()
+            table = self._hits.setdefault(route, {})
+            hits = [t for t in table.get(client, ()) if now - t < window]
+            if len(hits) >= limit:
+                table[client] = hits
+                return False, max(1, int(window - (now - hits[0])) + 1)
+            if client not in table and len(table) >= self.max_keys:
+                # At capacity, reclaim expired keys first...
+                self._reclaim_expired(route, now)
+                if len(table) >= self.max_keys:
+                    # ...and if every window is still live, this is an
+                    # identity flood, not organic load. Evicting here would
+                    # let an attacker reset their own limit on demand, so
+                    # refuse the NEW key instead. Fail closed: /v1/link is
+                    # invite-only and low-volume, so hitting this cap is an
+                    # attack, and turning tokens away beats minting them.
+                    LOG.error("rate limiter at capacity for %s (%d keys) with "
+                              "no expired windows: refusing new client %r",
+                              route, self.max_keys, client)
+                    return False, window
+            hits.append(now)
+            table[client] = hits
+            return True, 0
+
+    def _reclaim_expired(self, route, now):
+        """Caller holds the lock. Drop only keys whose window has fully
+        expired -- never a live one, or reclamation becomes the bypass."""
+        window = self.rules[route][1]
+        table = self._hits.get(route, {})
+        dead = [c for c, v in table.items() if not v or now - v[-1] >= window]
+        for c in dead:
+            del table[c]
+        if dead:
+            LOG.info("rate limiter reclaimed %d expired windows on %s",
+                     len(dead), route)
+
+
+def client_ip(handler, trust_forwarded_for, trusted_proxies=()):
+    """The address to rate-limit on.
+
+    Behind cloudflared every request arrives from the tunnel, so limiting on
+    the socket peer would let one abuser starve everyone. X-Forwarded-For is
+    client-controlled, though: a caller can prepend anything. Two guards:
+
+    1. The header is honored ONLY when the socket peer is itself a configured
+       trusted proxy. Without that, anyone who can reach the origin directly
+       -- and the origin also listens on the tailnet -- picks their own
+       rate-limit key and rotates it at will.
+    2. A trusted proxy APPENDS the peer it actually saw, so the LAST entry is
+       the only one the client did not choose. Take that, never the first.
+
+    The value must parse as a real IP; junk falls back to the socket peer
+    rather than becoming a key of its own.
+    """
+    peer = handler.client_address[0]
+    if not trust_forwarded_for:
+        return peer
+    if not _ip_in_any(peer, trusted_proxies):
+        LOG.warning("X-Forwarded-For ignored: peer %s is not a trusted proxy",
+                    peer)
+        return peer
+    xff = handler.headers.get("X-Forwarded-For", "")
+    parts = [p.strip() for p in xff.split(",") if p.strip()]
+    if not parts:
+        return peer
+    try:
+        return str(ipaddress.ip_address(parts[-1]))
+    except ValueError:
+        LOG.warning("X-Forwarded-For last hop %r is not an IP; using peer",
+                    parts[-1][:60])
+        return peer
+
+
+def _ip_in_any(addr, networks):
+    try:
+        ip = ipaddress.ip_address(addr)
+    except ValueError:
+        return False
+    for net in networks:
+        # Pre-parsed at startup by LinkService._parse_trusted_proxies, so a
+        # malformed entry can never reach here as a silent per-request skip.
+        if ip.version == net.version and ip in net:
+            return True
+    return False
+

 # --------------------------------------------------------------------------
 # HTTP helper (patchable in tests)
@@ -140,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.

@@ -187,6 +350,168 @@ class IdentityStore:
        data["identities"][str(sub)] = record
        self._write(data)

+    # -- device registry ---------------------------------------------------
+    # Devices hang off the identity that owns them, so "which computers can
+    # reach my files" has exactly one answer and it cannot drift from the
+    # identity map. Callers hold self.lock.
+
+    def record_device(self, sub, device_id, record):
+        data = self._load()
+        ident = data["identities"].get(str(sub))
+        if ident is None:
+            return False
+        devices = ident.setdefault("devices", {})
+        existing = devices.get(device_id, {})
+        # Re-linking the same device REPLACES its token name and clears any
+        # previous revocation: the user just proved identity again.
+        existing.update(record)
+        existing["revoked_at"] = None
+        devices[device_id] = existing
+        data["identities"][str(sub)] = ident
+        self._write(data)
+        return True
+
+    def devices_for_login(self, login):
+        """(sub, devices) for a Gitea login, or (None, {}).
+
+        Keyed on login because the caller authenticates with a forge token,
+        which proves a login -- not a Zitadel sub.
+        """
+        data = self._load()
+        for sub, ident in data["identities"].items():
+            if ident.get("login") == login:
+                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 {}
+        device = (ident.get("devices") or {}).get(device_id)
+        if device is None:
+            return False
+        device["revoked_at"] = when
+        self._write(data)
+        return True
+
+
+# --------------------------------------------------------------------------
+# Audit log: append-only JSONL, one line per security-relevant event
+# --------------------------------------------------------------------------
+
+class AuditLog:
+    """Who linked a computer, from where, when, and what was refused.
+
+    Append-only and separate from the identity map on purpose: state.json is
+    rewritten atomically on every change, so anything kept only there is one
+    bad write away from gone. An audit trail that can be rewritten by the
+    thing it audits is not an audit trail.
+
+    NOTE what this can and cannot see. Every event here is one granthi-link
+    handled. Git pushes and pulls do NOT pass through this service -- they go
+    straight to the forge -- so they are recorded by Gitea, not here. Reading
+    this file and believing it lists file activity would be a real mistake.
+    """
+
+    def __init__(self, path, keep_bytes=64 * 1024 * 1024):
+        self.path = path
+        self.keep_bytes = keep_bytes
+        self.lock = threading.Lock()
+
+    def write(self, event, **fields):
+        line = {"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+                "event": event}
+        line.update({k: v for k, v in fields.items() if v is not None})
+        try:
+            with self.lock:
+                self._rotate_if_needed()
+                fd = os.open(self.path,
+                             os.O_CREAT | os.O_WRONLY | os.O_APPEND, 0o600)
+                with os.fdopen(fd, "a") as f:
+                    f.write(json.dumps(line, sort_keys=True) + "\n")
+        except OSError as e:
+            # Never let auditing break the request it is auditing, but say so
+            # loudly -- a silently dead audit log is worse than none.
+            LOG.error("AUDIT WRITE FAILED (%s): %s", e, line)
+
+    def _rotate_if_needed(self):
+        try:
+            if os.path.getsize(self.path) < self.keep_bytes:
+                return
+        except OSError:
+            return
+        os.replace(self.path, self.path + ".1")
+
+    def read_for(self, login, limit=200):
+        """Events belonging to one login, newest first."""
+        out = []
+        for path in (self.path, self.path + ".1"):
+            try:
+                with open(path) as f:
+                    for raw in f:
+                        try:
+                            rec = json.loads(raw)
+                        except ValueError:
+                            continue
+                        if rec.get("login") == login:
+                            out.append(rec)
+            except OSError:
+                continue
+        out.sort(key=lambda r: r.get("ts") or "", reverse=True)
+        return out[:limit]
+

 # --------------------------------------------------------------------------
 # Core logic (class so tests can instantiate with a stub config)
@@ -202,6 +527,93 @@ class LinkService:
        self.userinfo_url = config.get(
            "zitadel_userinfo", "https://id.shre.ai/oidc/v1/userinfo")
        self.state = IdentityStore(config.get("state_path", DEFAULT_STATE))
+        self.audit = AuditLog(config.get("audit_path", DEFAULT_AUDIT))
+
+        # Rate limiting is ON by default: this endpoint creates accounts and
+        # mints tokens, so the safe default is limited, and disabling it has
+        # to be a deliberate config act rather than an omission.
+        rl = config.get("rate_limit", {})
+        if rl is None:
+            rl = {}
+        if not isinstance(rl, dict):
+            raise SystemExit("config rate_limit must be an object")
+        # Real JSON booleans only. `"enabled": null` or `0` must not quietly
+        # turn limiting off, and the string "false" must not turn XFF trust
+        # ON (every non-empty string is truthy).
+        for flag, default in (("enabled", True), ("trust_forwarded_for", False)):
+            if flag in rl and not isinstance(rl[flag], bool):
+                raise SystemExit(
+                    f"config rate_limit.{flag} must be true or false, "
+                    f"got {rl[flag]!r}")
+        self.trust_forwarded_for = rl.get("trust_forwarded_for", False)
+        self.trusted_proxies = self._parse_trusted_proxies(
+            rl.get("trusted_proxies"))
+        if self.trust_forwarded_for and not self.trusted_proxies:
+            raise SystemExit(
+                "config rate_limit.trust_forwarded_for requires a non-empty "
+                "trusted_proxies list -- trusting the header from any peer "
+                "lets callers choose their own rate-limit key")
+        if rl.get("enabled", True):
+            rules = dict(DEFAULT_RATE_RULES)
+            for route, spec in (rl.get("rules") or {}).items():
+                # `type(x) is int`, NOT isinstance: bool subclasses int, so
+                # isinstance lets [5, true] through as a 1-SECOND window --
+                # an hourly limit silently becomes ~5/sec -- and false in the
+                # limit slot disables the endpoint.
+                if (not isinstance(spec, (list, tuple)) or len(spec) != 2
+                        or not all(type(x) is int for x in spec)
+                        or spec[1] <= 0):
+                    # A malformed rule must not silently mean "unlimited".
+                    raise SystemExit(
+                        f"config rate_limit.rules[{route!r}] must be "
+                        f"[max_requests, window_seconds] with window > 0")
+                rules[route] = (spec[0], spec[1])
+            self.limiter = RateLimiter(rules)
+            LOG.info("rate limiting active: %s (trust_forwarded_for=%s, "
+                     "trusted_proxies=%s)",
+                     {k: f"{v[0]}/{v[1]}s" for k, v in rules.items()},
+                     self.trust_forwarded_for, list(self.trusted_proxies))
+        else:
+            self.limiter = None
+            LOG.warning("rate limiting DISABLED by config -- /v1/link will "
+                        "mint tokens without any throttle")
+
+    @staticmethod
+    def _parse_trusted_proxies(raw):
+        """Validate at STARTUP, not per request.
+
+        This setting gates a spoofable identity, so every failure mode has to
+        be loud and early: a bare string would be iterated character by
+        character (each char a "network"), malformed entries would only
+        surface as a per-request log line, and a wildcard like 0.0.0.0/0 or
+        ::/0 quietly restores "trust X-Forwarded-For from anyone" -- the exact
+        hole trusted_proxies exists to close.
+        """
+        if raw is None:
+            return ()
+        if isinstance(raw, str) or not isinstance(raw, (list, tuple)):
+            raise SystemExit(
+                "config rate_limit.trusted_proxies must be a list of CIDRs, "
+                f"got {type(raw).__name__}")
+        nets = []
+        for entry in raw:
+            if not isinstance(entry, str):
+                raise SystemExit(
+                    f"config rate_limit.trusted_proxies entry {entry!r} "
+                    "must be a string")
+            try:
+                net = ipaddress.ip_network(entry, strict=False)
+            except ValueError as e:
+                raise SystemExit(
+                    f"config rate_limit.trusted_proxies entry {entry!r} "
+                    f"is not a valid network: {e}")
+            if net.prefixlen == 0:
+                raise SystemExit(
+                    f"config rate_limit.trusted_proxies entry {entry!r} "
+                    "matches every address, which is the same as trusting "
+                    "X-Forwarded-For from any peer -- refusing")
+            nets.append(net)
+        return tuple(nets)

    def test_mode_enabled(self):
        """Config test_mode is honored ONLY with the env gate also set."""
@@ -276,16 +688,32 @@ 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

-    def mint_token(self, login, device_name):
+    def mint_token(self, login, device_name, device_id=None):
        """Admin basic auth + Sudo header. Verified on Gitea 1.27.1:
-        token-auth sudo (header or ?sudo=) -> 401; basic+Sudo -> 201."""
+        token-auth sudo (header or ?sudo=) -> 401; basic+Sudo -> 201.
+
+        The device id is part of the token NAME because revocation deletes by
+        name. With only device_name + a one-second timestamp, one user linking
+        two machines called "macbook" in the same second would collide -- and
+        a collision means signing out one laptop kills the other's access.
+        """
        safe_dev = LOGIN_SAFE.sub("-", device_name or "device")[:40]
-        token_name = "granthi-sync-{}-{}".format(
-            safe_dev, datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"))
+        suffix = LOGIN_SAFE.sub("-", str(device_id or ""))[:12]
+        token_name = "granthi-sync-{}-{}{}".format(
+            safe_dev, datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"),
+            f"-{suffix}" if suffix else "")
        status, resp = http_json(
            "POST", f"{self.gitea}/api/v1/users/{login}/tokens",
            headers={
@@ -324,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)
@@ -349,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:
@@ -375,7 +807,7 @@ class LinkService:
                              "identity"}, None

    # -- endpoints ---------------------------------------------------------
-    def link(self, body):
+    def link(self, body, client_ip=None):
        device_name = body.get("device_name") or "device"
        if self.test_mode_enabled() and isinstance(body.get("test_userinfo"), dict):
            LOG.warning("TEST-MODE link request (stubbed userinfo)")
@@ -398,14 +830,300 @@ class LinkService:
            return 500, {"error": "identity state unavailable"}
        if login is None:
            return status, err_resp
-        gitea_token, token_name, err = self.mint_token(login, device_name)
+        wanted_device = str(body.get("device_id") or "").strip() or None
+        gitea_token, token_name, err = self.mint_token(login, device_name,
+                                                       wanted_device)
        if err:
            return 502, {"error": err}
        LOG.info("minted token %s for %s (sub %s)", token_name, login, sub)
-        return 200, {"gitea_base": self.public_gitea, "login": login,
-                     "token": gitea_token, "token_name": token_name}
+        # A device id the client generated once and keeps. Clients older than
+        # v1.2 do not send one; fall back to the token name, which is unique
+        # per link, so every device still lands in the registry.
+        device_id = wanted_device or token_name
+        with self.state.lock:
+            self.state.record_device(sub, device_id, {
+                "name": device_name,
+                "token_name": token_name,
+                "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)
+        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

-    def repos(self, body):
+    # -- device registry endpoints -----------------------------------------
+
+    def whoami(self, token):
+        """The login a forge token belongs to, or None.
+
+        Authorisation for every device endpoint rests on this: the caller
+        proves who they are by holding a working token for that account, and
+        the forge is the one that decides. No login is ever taken from the
+        request body.
+        """
+        if not token or not isinstance(token, str):
+            return None
+        status, resp = http_json(
+            "GET", f"{self.gitea}/api/v1/user",
+            headers={"Authorization": f"token {token}"})
+        if status == 200 and isinstance(resp, dict):
+            return resp.get("login")
+        return None
+
+    def devices(self, body, client_ip=None):
+        login = self.whoami(body.get("token"))
+        if not login:
+            return 401, {"error": "invalid or revoked token"}
+        with self.state.lock:
+            _, devices = self.state.devices_for_login(login)
+        out = []
+        for device_id, rec in sorted(
+                devices.items(), key=lambda kv: kv[1].get("linked_at") or ""):
+            out.append({"device_id": device_id,
+                        "name": rec.get("name") or "device",
+                        "linked_at": rec.get("linked_at"),
+                        "token_name": rec.get("token_name"),
+                        "revoked_at": rec.get("revoked_at")})
+        return 200, {"login": login, "devices": out}
+
+    def revoke_device(self, body, client_ip=None):
+        """Sign one computer out. Immediate: the forge token is deleted, so
+        the next fetch or push from that machine fails at the server. There
+        is no 'stop syncing' flag for a client to honour or ignore."""
+        login = self.whoami(body.get("token"))
+        if not login:
+            return 401, {"error": "invalid or revoked token"}
+        device_id = str(body.get("device_id") or "").strip()
+        if not device_id:
+            return 400, {"error": "device_id required"}
+        with self.state.lock:
+            sub, devices = self.state.devices_for_login(login)
+            record = devices.get(device_id)
+            if sub is None or record is None:
+                # Do not confirm the existence of ids on other accounts.
+                self.audit.write("device.revoke.denied", login=login,
+                                 device_id=device_id,
+                                 client_ip=client_ip,
+                                 reason="not a device of this account")
+                return 404, {"error": "no such device on this account"}
+            token_name = record.get("token_name")
+            err = self.delete_forge_token(login, token_name)
+            if err:
+                self.audit.write("device.revoke.failed", login=login,
+                                 device_id=device_id, token_name=token_name,
+                                 client_ip=client_ip, reason=err)
+                # Do NOT mark it revoked: the token still works, and a
+                # registry claiming otherwise is worse than an honest error.
+                return 502, {"error": err}
+            when = datetime.now(timezone.utc).isoformat(timespec="seconds")
+            self.state.mark_revoked(sub, device_id, when)
+        self.audit.write("device.revoke", login=login, device_id=device_id,
+                         token_name=token_name, client_ip=client_ip)
+        LOG.info("revoked device %s (%s) for %s", device_id, token_name, login)
+        return 200, {"revoked": device_id, "revoked_at": when}
+
+    def delete_forge_token(self, login, token_name):
+        """Gitea refuses token-auth deletion of tokens (403); admin basic
+        auth + Sudo works (verified 204 on 1.27.2)."""
+        if not token_name:
+            return "device has no recorded token to revoke"
+        status, resp = http_json(
+            "DELETE",
+            f"{self.gitea}/api/v1/users/{login}/tokens/"
+            f"{urllib.parse.quote(token_name, safe='')}",
+            headers={"Authorization": _basic(self.cfg["admin_login"],
+                                             self.cfg["admin_password"]),
+                     "Sudo": login})
+        if status in (204, 404):
+            # 404 = already gone. The caller wanted it dead; it is dead.
+            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:
+            return 401, {"error": "invalid or revoked token"}
+        try:
+            limit = min(int(body.get("limit") or 100), 500)
+        except (TypeError, ValueError):
+            limit = 100
+        return 200, {"login": login,
+                     "events": self.audit.read_for(login, limit=limit),
+                     "note": "granthi-link events only; git pushes and pulls "
+                             "go straight to the forge and are recorded "
+                             "there, not here"}
+
+    def repos(self, body, client_ip=None):
        token = body.get("token")
        name = body.get("name")
        if not token or not name:
@@ -452,6 +1170,29 @@ class Handler(BaseHTTPRequestHandler):
            self._send(404, {"error": "not found"})

    def do_POST(self):
+        # Rate-limit BEFORE reading the body or doing any work -- the point is
+        # to spend nothing on an abusive caller. Same close-the-connection
+        # treatment the 413 path uses, for the same reason: we are not going
+        # to drain a body we already decided to reject.
+        limiter = getattr(self.service, "limiter", None)
+        if limiter is not None:
+            who = client_ip(self, self.service.trust_forwarded_for,
+                            self.service.trusted_proxies)
+            allowed, retry_after = limiter.check(self.path, who)
+            if not allowed:
+                LOG.warning("rate limited %s %s (retry after %ss)",
+                            who, self.path, retry_after)
+                self.close_connection = True
+                payload = json.dumps({
+                    "error": "rate limit exceeded",
+                    "retry_after": retry_after}).encode()
+                self.send_response(429)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Retry-After", str(retry_after))
+                self.send_header("Content-Length", str(len(payload)))
+                self.end_headers()
+                self.wfile.write(payload)
+                return
        cl = self.headers.get("Content-Length")
        if cl is None:
            self.close_connection = True
@@ -475,10 +1216,25 @@ class Handler(BaseHTTPRequestHandler):
            return self._send(400, {"error": "invalid JSON body"})
        if not isinstance(body, dict):
            return self._send(400, {"error": "body must be a JSON object"})
+        # Recomputed rather than reused from the rate-limit block above:
+        # that block is skipped entirely when no limiter is configured, and
+        # an audit trail must not have holes because limiting was off.
+        peer = client_ip(self, self.service.trust_forwarded_for,
+                         self.service.trusted_proxies)
        if self.path == "/v1/link":
-            status, resp = self.service.link(body)
+            status, resp = self.service.link(body, peer)
        elif self.path == "/v1/repos":
-            status, resp = self.service.repos(body)
+            status, resp = self.service.repos(body, peer)
+        elif self.path == "/v1/devices":
+            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:
            return self._send(404, {"error": "not found"})
        self._send(status, resp)
@@ -9,6 +9,7 @@ import sys
 import tempfile
 import threading
 import unittest
+import urllib.parse
 import urllib.request
 from unittest import mock
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -43,6 +44,21 @@ class StubUpstream(BaseHTTPRequestHandler):
                                        "[email protected]", "email":
                                        "[email protected]", "name": "Alice"})
            return self._json(401, {"error": "invalid token"})
+        if self.path == "/api/v1/user":
+            # whoami: the forge decides who a token belongs to
+            tok = self.headers.get("Authorization", "").replace("token ", "")
+            login = st["tokens_by_sha"].get(tok)
+            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"]:
@@ -61,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"]})
@@ -71,7 +91,14 @@ class StubUpstream(BaseHTTPRequestHandler):
                "sudo": self.headers.get("Sudo", ""), "body": body})
            if not self.headers.get("Authorization", "").startswith("Basic "):
                return self._json(401, {"message": "auth required"})
-            return self._json(201, {"sha1": "MINTED", "name": body["name"]})
+            login = self.path.split("/")[4]
+            # unique per mint, like a real forge -- two users may legitimately
+            # hold the same token NAME
+            st["sha_seq"] = st.get("sha_seq", 0) + 1
+            sha = f"SHA-{login}-{st['sha_seq']}-{body['name']}"
+            st["tokens_by_sha"][sha] = login
+            st["tokens"].setdefault(login, set()).add(body["name"])
+            return self._json(201, {"sha1": sha, "name": body["name"]})
        if self.path == "/api/v1/user/repos":
            if body["name"] in st["repos"]:
                return self._json(409, {"message": "exists"})
@@ -81,6 +108,54 @@ 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
+                return self._json(403, {"message": "basic auth required"})
+            login, name = self.path.split("/tokens/")
+            login = login.rsplit("/", 1)[1]
+            name = urllib.parse.unquote(name)
+            if st.get("revoke_fails"):
+                return self._json(500, {"message": "forge exploded"})
+            if name not in st["tokens"].get(login, set()):
+                return self._json(404, {"message": "not found"})
+            st["tokens"][login].discard(name)
+            for sha, owner in list(st["tokens_by_sha"].items()):
+                if owner == login and sha.endswith(f"-{name}"):
+                    del st["tokens_by_sha"][sha]
+            return self._json(204, {})
+        self._json(404, {})
+
    def log_message(self, *a):
        pass

@@ -88,7 +163,10 @@ class StubUpstream(BaseHTTPRequestHandler):
 class ServiceTestBase(unittest.TestCase):
    def setUp(self):
        StubUpstream.state = {"users": {}, "created": [], "repos": set(),
-                              "token_reqs": [], "hide_once": set()}
+                              "token_reqs": [], "hide_once": set(),
+                              "tokens": {}, "tokens_by_sha": {},
+                              "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)
@@ -102,6 +180,7 @@ class ServiceTestBase(unittest.TestCase):
            "admin_token": "ADMTOK", "admin_login": "root",
            "admin_password": "rootpw", "test_mode": False,
            "state_path": self.state_path,
+            "audit_path": os.path.join(self.state_dir, "audit.jsonl"),
        })

    def enable_test_mode(self):
@@ -111,13 +190,17 @@ class ServiceTestBase(unittest.TestCase):
        patcher.start()
        self.addCleanup(patcher.stop)

-    def stub_link(self, sub, username, email=None, verified=None, device="d"):
+    def stub_link(self, sub, username, email=None, verified=None, device="d",
+                  device_id=None, client_ip=None):
        ui = {"sub": sub, "preferred_username": username}
        if email is not None:
            ui["email"] = email
        if verified is not None:
            ui["email_verified"] = verified
-        return self.svc.link({"test_userinfo": ui, "device_name": device})
+        body = {"test_userinfo": ui, "device_name": device}
+        if device_id:
+            body["device_id"] = device_id
+        return self.svc.link(body, client_ip)

    def read_state(self):
        with open(self.state_path) as f:
@@ -146,7 +229,8 @@ class TestLink(ServiceTestBase):
                                      "device_name": "mac studio"})
        self.assertEqual(status, 200)
        self.assertEqual(resp["login"], "alice.smith")
-        self.assertEqual(resp["token"], "MINTED")
+        # the stub now issues a per-token sha so revocation can be tested
+        self.assertIn("granthi-sync-", resp["token"])
        self.assertEqual(resp["gitea_base"], "http://public.example:3041")
        st = StubUpstream.state
        self.assertEqual(len(st["created"]), 1)
@@ -480,5 +564,843 @@ class TestHealth(HandlerTestBase):
        self.assertEqual(body["service"], "granthi-link")


+
+class TestDeviceRegistry(ServiceTestBase):
+    """Which computers can reach my files, and can I cut one off."""
+
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+
+    def link_device(self, device_id, name="laptop"):
+        status, resp = self.stub_link("s1", "alice", device=name,
+                                      device_id=device_id)
+        self.assertEqual(status, 200, resp)
+        return resp
+
+    def test_devices_are_recorded_and_listed_per_account(self):
+        self.link_device("dev-a", "work-laptop")
+        second = self.link_device("dev-b", "home-mac")
+        status, resp = self.svc.devices({"token": second["token"]})
+        self.assertEqual(status, 200)
+        self.assertEqual(resp["login"], "alice")
+        ids = [d["device_id"] for d in resp["devices"]]
+        self.assertEqual(sorted(ids), ["dev-a", "dev-b"])
+        names = {d["device_id"]: d["name"] for d in resp["devices"]}
+        self.assertEqual(names["dev-a"], "work-laptop")
+
+    def test_listing_needs_a_working_token(self):
+        self.link_device("dev-a")
+        status, _ = self.svc.devices({"token": "not-a-real-token"})
+        self.assertEqual(status, 401)
+        status, _ = self.svc.devices({})
+        self.assertEqual(status, 401)
+
+    def test_login_is_never_taken_from_the_request_body(self):
+        """Authorisation comes from the forge's answer about the token, so a
+        body claiming another account changes nothing."""
+        first = self.link_device("dev-a")
+        status, resp = self.svc.devices({"token": first["token"],
+                                         "login": "somebody-else"})
+        self.assertEqual(resp["login"], "alice")
+
+    def test_relinking_the_same_device_does_not_duplicate_it(self):
+        self.link_device("dev-a", "laptop")
+        self.link_device("dev-a", "laptop-renamed")
+        status, resp = self.svc.devices({"token": self.link_device("dev-a")["token"]})
+        self.assertEqual(len(resp["devices"]), 1)
+
+    def test_old_clients_without_a_device_id_still_register(self):
+        resp = self.stub_link("s1", "alice", device="ancient")[1]
+        self.assertEqual(resp["device_id"], resp["token_name"])
+        status, listed = self.svc.devices({"token": resp["token"]})
+        self.assertEqual([d["device_id"] for d in listed["devices"]],
+                         [resp["token_name"]])
+
+
+class TestRevoke(ServiceTestBase):
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+        self.a = self.stub_link("s1", "alice", device="laptop-a",
+                                device_id="dev-a")[1]
+        self.b = self.stub_link("s1", "alice", device="laptop-b",
+                                device_id="dev-b")[1]
+
+    def test_revoking_a_device_kills_its_token_at_the_forge(self):
+        status, resp = self.svc.revoke_device({"token": self.b["token"],
+                                               "device_id": "dev-a"})
+        self.assertEqual(status, 200, resp)
+        # the revoked device's token no longer authenticates ANYTHING
+        self.assertEqual(self.svc.whoami(self.a["token"]), None)
+        # the device that did the revoking still works
+        self.assertEqual(self.svc.whoami(self.b["token"]), "alice")
+
+    def test_a_device_can_revoke_itself(self):
+        status, _ = self.svc.revoke_device({"token": self.a["token"],
+                                            "device_id": "dev-a"})
+        self.assertEqual(status, 200)
+        self.assertEqual(self.svc.whoami(self.a["token"]), None)
+
+    def test_revoked_device_is_marked_not_deleted(self):
+        self.svc.revoke_device({"token": self.b["token"],
+                                "device_id": "dev-a"})
+        status, resp = self.svc.devices({"token": self.b["token"]})
+        by_id = {d["device_id"]: d for d in resp["devices"]}
+        self.assertIsNotNone(by_id["dev-a"]["revoked_at"])
+        self.assertIsNone(by_id["dev-b"]["revoked_at"])
+
+    def test_cannot_revoke_a_device_on_another_account(self):
+        other = self.stub_link("s2", "mallory", device="theirs",
+                               device_id="dev-x")[1]
+        status, resp = self.svc.revoke_device({"token": other["token"],
+                                               "device_id": "dev-a"})
+        self.assertEqual(status, 404)
+        # alice's device is untouched
+        self.assertEqual(self.svc.whoami(self.a["token"]), "alice")
+
+    def test_a_failed_forge_delete_is_not_recorded_as_revoked(self):
+        """A registry that says 'revoked' while the token still works is
+        worse than an honest error."""
+        StubUpstream.state["revoke_fails"] = True
+        status, _ = self.svc.revoke_device({"token": self.b["token"],
+                                            "device_id": "dev-a"})
+        self.assertEqual(status, 502)
+        StubUpstream.state["revoke_fails"] = False
+        _, listed = self.svc.devices({"token": self.b["token"]})
+        by_id = {d["device_id"]: d for d in listed["devices"]}
+        self.assertIsNone(by_id["dev-a"]["revoked_at"])
+        self.assertEqual(self.svc.whoami(self.a["token"]), "alice")
+
+    def test_revoking_twice_is_not_an_error(self):
+        self.svc.revoke_device({"token": self.b["token"], "device_id": "dev-a"})
+        status, _ = self.svc.revoke_device({"token": self.b["token"],
+                                            "device_id": "dev-a"})
+        self.assertEqual(status, 200)  # forge 404 = already gone = success
+
+
+class TestAuditLog(ServiceTestBase):
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+
+    def test_link_and_revoke_are_recorded_with_who_and_where(self):
+        first = self.stub_link("s1", "alice", device="laptop",
+                               device_id="dev-a", client_ip="203.0.113.9")[1]
+        self.stub_link("s1", "alice", device="mac", device_id="dev-b")
+        second = self.svc.devices({"token": first["token"]})[1]
+        self.assertEqual(len(second["devices"]), 2)
+        self.svc.revoke_device({"token": first["token"],
+                                "device_id": "dev-b"}, client_ip="198.51.100.4")
+
+        status, resp = self.svc.audit_read({"token": first["token"]})
+        self.assertEqual(status, 200)
+        events = resp["events"]
+        kinds = [e["event"] for e in events]
+        self.assertIn("device.link", kinds)
+        self.assertIn("device.revoke", kinds)
+        link_ev = [e for e in events if e["event"] == "device.link"
+                   and e["device_id"] == "dev-a"][0]
+        self.assertEqual(link_ev["client_ip"], "203.0.113.9")
+        self.assertEqual(link_ev["login"], "alice")
+        revoke_ev = [e for e in events if e["event"] == "device.revoke"][0]
+        self.assertEqual(revoke_ev["client_ip"], "198.51.100.4")
+        self.assertEqual(revoke_ev["device_id"], "dev-b")
+
+    def test_events_are_newest_first_and_scoped_to_the_caller(self):
+        mine = self.stub_link("s1", "alice", device_id="dev-a")[1]
+        self.stub_link("s2", "mallory", device_id="dev-x")
+        _, resp = self.svc.audit_read({"token": mine["token"]})
+        self.assertTrue(resp["events"])
+        self.assertTrue(all(e["login"] == "alice" for e in resp["events"]))
+        stamps = [e["ts"] for e in resp["events"]]
+        self.assertEqual(stamps, sorted(stamps, reverse=True))
+
+    def test_reading_the_log_needs_a_working_token(self):
+        self.stub_link("s1", "alice", device_id="dev-a")
+        status, _ = self.svc.audit_read({"token": "nope"})
+        self.assertEqual(status, 401)
+
+    def test_a_refused_revoke_is_recorded_too(self):
+        self.stub_link("s1", "alice", device_id="dev-a")
+        other = self.stub_link("s2", "mallory", device_id="dev-x")[1]
+        self.svc.revoke_device({"token": other["token"],
+                                "device_id": "dev-a"}, client_ip="192.0.2.7")
+        _, resp = self.svc.audit_read({"token": other["token"]})
+        denied = [e for e in resp["events"]
+                  if e["event"] == "device.revoke.denied"]
+        self.assertEqual(len(denied), 1)
+        self.assertEqual(denied[0]["client_ip"], "192.0.2.7")
+
+    def test_the_log_says_what_it_cannot_see(self):
+        """Reading this and believing it lists file activity would be a real
+        mistake: git traffic never passes through this service."""
+        mine = self.stub_link("s1", "alice", device_id="dev-a")[1]
+        _, resp = self.svc.audit_read({"token": mine["token"]})
+        self.assertIn("pushes and pulls", resp["note"])
+
+    def test_a_broken_audit_file_does_not_break_the_request(self):
+        self.svc.audit.path = "/nonexistent-dir/audit.jsonl"
+        status, _ = self.stub_link("s1", "alice", device_id="dev-a")
+        self.assertEqual(status, 200)  # linking still works
+
+    def test_rotation_keeps_the_previous_file_readable(self):
+        self.svc.audit.keep_bytes = 1500  # one rotation, not many
+        mine = self.stub_link("s1", "alice", device_id="dev-a")[1]
+        for i in range(20):
+            self.svc.audit.write("noise", login="alice", n=i)
+        _, resp = self.svc.audit_read({"token": mine["token"]}, )
+        self.assertGreater(len(resp["events"]), 15)  # spans both files
+        self.assertTrue(os.path.exists(self.svc.audit.path + ".1"))
+
+
+class TestTokenNameUniqueness(ServiceTestBase):
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+
+    def test_two_devices_named_the_same_get_different_token_names(self):
+        """Revocation deletes by token name, so a collision would mean
+        signing out one laptop kills the other."""
+        a = self.stub_link("s1", "alice", device="macbook", device_id="dev-a")[1]
+        b = self.stub_link("s1", "alice", device="macbook", device_id="dev-b")[1]
+        self.assertNotEqual(a["token_name"], b["token_name"])
+        self.svc.revoke_device({"token": b["token"], "device_id": "dev-a"})
+        self.assertIsNone(self.svc.whoami(a["token"]))
+        self.assertEqual(self.svc.whoami(b["token"]), "alice")
+
+
+class TestDeviceEndpointsOverHttp(HandlerTestBase):
+    """Through the real handler, not the service method.
+
+    The first live run of the revoke endpoint returned 429 on every call
+    while every unit test passed: the tests called svc.revoke_device()
+    directly, so nothing ever went through the rate limiter. Any route whose
+    limit is a policy decision needs at least one test that speaks HTTP.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+        self.a = self.stub_link("s1", "alice", device="laptop",
+                                device_id="dev-a")[1]
+        self.b = self.stub_link("s1", "alice", device="desktop",
+                                device_id="dev-b")[1]
+
+    def post(self, path, obj):
+        body = json.dumps(obj).encode()
+        return self.raw_post(path, body,
+                             {"Content-Type": "application/json",
+                              "Content-Length": str(len(body))})
+
+    def test_revoke_is_reachable_and_not_rate_limited_away(self):
+        status, resp = self.post("/v1/devices/revoke",
+                                 {"token": self.b["token"],
+                                  "device_id": "dev-a"})
+        self.assertEqual(status, 200, resp)
+        self.assertIsNone(self.svc.whoami(self.a["token"]))
+
+    def test_repeated_revokes_keep_working(self):
+        """A person signing several lost machines out in one sitting must not
+        be locked out partway through."""
+        for i in range(12):
+            self.stub_link("s1", "alice", device=f"d{i}", device_id=f"gone-{i}")
+        for i in range(12):
+            status, resp = self.post("/v1/devices/revoke",
+                                     {"token": self.b["token"],
+                                      "device_id": f"gone-{i}"})
+            self.assertEqual(status, 200, f"revoke {i}: {resp}")
+
+    def test_devices_and_audit_are_reachable_over_http(self):
+        status, resp = self.post("/v1/devices", {"token": self.a["token"]})
+        self.assertEqual(status, 200)
+        self.assertEqual(len(resp["devices"]), 2)
+        status, resp = self.post("/v1/audit", {"token": self.a["token"]})
+        self.assertEqual(status, 200)
+        self.assertTrue(resp["events"])
+
+    def test_no_route_a_client_uses_is_configured_to_zero(self):
+        """0 means DISABLED in this limiter, so a zero on a live route is a
+        dead endpoint, not an unlimited one."""
+        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()
+
+
+class TestRateLimiterUnit(unittest.TestCase):
+    """The limiter in isolation, on a fake clock -- no sleeping in tests."""
+
+    def setUp(self):
+        self.now = 1000.0
+        self.rl = granthi_link.RateLimiter(
+            rules={"/v1/link": (3, 60)}, clock=lambda: self.now)
+
+    def test_allows_up_to_the_limit_then_denies(self):
+        for i in range(3):
+            allowed, _ = self.rl.check("/v1/link", "1.1.1.1")
+            self.assertTrue(allowed, f"request {i} should pass")
+        allowed, retry = self.rl.check("/v1/link", "1.1.1.1")
+        self.assertFalse(allowed)
+        self.assertGreater(retry, 0)
+        self.assertLessEqual(retry, 61)
+
+    def test_window_slides(self):
+        for _ in range(3):
+            self.rl.check("/v1/link", "1.1.1.1")
+        self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
+        self.now += 61
+        self.assertTrue(self.rl.check("/v1/link", "1.1.1.1")[0])
+
+    def test_denied_requests_do_not_extend_the_window(self):
+        """A client that keeps hammering must not push its own window
+        forward and lock itself out forever."""
+        for _ in range(3):
+            self.rl.check("/v1/link", "1.1.1.1")
+        for _ in range(20):          # hammer while denied
+            self.now += 1
+            self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
+        self.now = 1000.0 + 61       # just past the ORIGINAL window
+        self.assertTrue(self.rl.check("/v1/link", "1.1.1.1")[0])
+
+    def test_clients_are_isolated(self):
+        for _ in range(3):
+            self.rl.check("/v1/link", "1.1.1.1")
+        self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
+        self.assertTrue(self.rl.check("/v1/link", "2.2.2.2")[0])
+
+    def test_routes_are_isolated_and_unknown_routes_pass(self):
+        for _ in range(3):
+            self.rl.check("/v1/link", "1.1.1.1")
+        self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
+        self.assertTrue(self.rl.check("/v1/repos", "1.1.1.1")[0])
+        for _ in range(50):
+            self.assertTrue(self.rl.check("/health", "1.1.1.1")[0])
+
+    def test_zero_limit_disables_the_endpoint(self):
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (0, 60)},
+                                      clock=lambda: self.now)
+        self.assertFalse(rl.check("/v1/link", "1.1.1.1")[0])
+
+    def test_key_store_stays_bounded(self):
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (5, 60)},
+                                      max_keys=50, clock=lambda: self.now)
+        for i in range(500):
+            self.now += 0.001
+            rl.check("/v1/link", f"10.0.{i // 256}.{i % 256}")
+        self.assertLessEqual(len(rl._hits["/v1/link"]), 50 + 1)
+
+    def test_expired_keys_are_reclaimed(self):
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (5, 60)},
+                                      max_keys=10, clock=lambda: self.now)
+        for i in range(10):
+            rl.check("/v1/link", f"10.0.0.{i}")
+        self.now += 120                      # everything expires
+        for i in range(10, 25):
+            rl.check("/v1/link", f"10.0.0.{i}")
+        self.assertLessEqual(len(rl._hits["/v1/link"]), 11)
+
+    def test_concurrent_checks_never_exceed_the_limit(self):
+        """The lock has to actually hold under threads: 40 racing callers
+        against a limit of 10 must yield exactly 10 allows."""
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (10, 60)})
+        results, lock = [], threading.Lock()
+
+        def hit():
+            a, _ = rl.check("/v1/link", "9.9.9.9")
+            with lock:
+                results.append(a)
+
+        ts = [threading.Thread(target=hit) for _ in range(40)]
+        for t in ts:
+            t.start()
+        for t in ts:
+            t.join()
+        self.assertEqual(sum(results), 10)
+
+
+class _FakeHandler:
+    def __init__(self, peer, xff=None):
+        self.client_address = (peer, 12345)
+        self.headers = {} if xff is None else {"X-Forwarded-For": xff}
+        if xff is not None:
+            self.headers = type("H", (), {"get": lambda s, k, d="":
+                                          xff if k == "X-Forwarded-For" else d})()
+
+
+class TestClientIp(unittest.TestCase):
+    # Pre-parsed, exactly as LinkService._parse_trusted_proxies hands them
+    # over -- validation happens once at startup, never per request.
+    PROXY = granthi_link.LinkService._parse_trusted_proxies(["10.0.0.0/8"])
+
+    def test_socket_peer_by_default(self):
+        h = _FakeHandler("5.5.5.5", xff="1.2.3.4")
+        self.assertEqual(granthi_link.client_ip(h, False, self.PROXY), "5.5.5.5")
+
+    def test_trusted_proxy_uses_the_last_hop_not_the_client_supplied_first(self):
+        """A caller can PREPEND anything to X-Forwarded-For; a trusted proxy
+        appends the peer it really saw. Only the last entry is trustworthy."""
+        h = _FakeHandler("10.0.0.1", xff="1.2.3.4, 203.0.113.9")
+        self.assertEqual(granthi_link.client_ip(h, True, self.PROXY),
+                         "203.0.113.9")
+
+    def test_untrusted_peer_cannot_choose_its_own_key(self):
+        """The origin also listens on the tailnet. Anyone reaching it
+        directly must not be able to pick (and rotate) their rate-limit key
+        just by sending a header."""
+        h = _FakeHandler("203.0.113.50", xff="9.9.9.9")
+        self.assertEqual(granthi_link.client_ip(h, True, self.PROXY),
+                         "203.0.113.50")
+
+    def test_non_ip_last_hop_falls_back_to_peer(self):
+        for junk in ("not-an-ip", "', OR 1=1", "x" * 500, "10.0.0.1:8080"):
+            with self.subTest(junk=junk):
+                h = _FakeHandler("10.0.0.1", xff=f"1.2.3.4, {junk}")
+                self.assertEqual(
+                    granthi_link.client_ip(h, True, self.PROXY), "10.0.0.1")
+
+    def test_trusted_proxy_falls_back_when_header_absent(self):
+        h = _FakeHandler("10.0.0.1")
+        self.assertEqual(granthi_link.client_ip(h, True, self.PROXY),
+                         "10.0.0.1")
+
+    def test_malformed_trusted_proxy_entry_is_rejected_at_startup(self):
+        """It never reaches a request: a bad CIDR kills the process rather
+        than degrading to a per-request log line."""
+        with self.assertRaises(SystemExit):
+            granthi_link.LinkService._parse_trusted_proxies(["not-a-cidr"])
+
+    def test_v4_peer_does_not_match_a_v6_trusted_net(self):
+        nets = granthi_link.LinkService._parse_trusted_proxies(["fd00::/8"])
+        h = _FakeHandler("10.0.0.1", xff="9.9.9.9")
+        self.assertEqual(granthi_link.client_ip(h, True, nets), "10.0.0.1")
+
+
+class TestRateLimitConfig(ServiceTestBase):
+    def _svc(self, rl):
+        cfg = dict(self.svc.cfg)
+        cfg["rate_limit"] = rl
+        return granthi_link.LinkService(cfg)
+
+    def test_enabled_by_default_when_key_absent(self):
+        self.assertIsNotNone(self.svc.limiter)
+
+    def test_explicit_disable_is_honored(self):
+        self.assertIsNone(self._svc({"enabled": False}).limiter)
+
+    def test_custom_rule_overrides_default(self):
+        svc = self._svc({"rules": {"/v1/link": [99, 120]}})
+        self.assertEqual(svc.limiter.rules["/v1/link"], (99, 120))
+
+    def test_malformed_rule_refuses_startup_rather_than_meaning_unlimited(self):
+        for bad in ({"rules": {"/v1/link": [5]}},
+                    {"rules": {"/v1/link": "5/hour"}},
+                    {"rules": {"/v1/link": [5, 0]}},
+                    {"rules": {"/v1/link": [5, -1]}},
+                    {"rules": {"/v1/link": ["5", "60"]}}):
+            with self.subTest(cfg=bad):
+                with self.assertRaises(SystemExit):
+                    self._svc(bad)
+
+
+class TestRateLimitOverHttp(HandlerTestBase):
+    """Proven on the wire: real 429, real Retry-After."""
+
+    def setUp(self):
+        super().setUp()
+        self.svc.limiter = granthi_link.RateLimiter(rules={"/v1/link": (2, 60)})
+
+    def test_third_request_gets_429_with_retry_after(self):
+        body = json.dumps({"zitadel_access_token": "x"}).encode()
+        hdrs = {"Content-Length": str(len(body)),
+                "Content-Type": "application/json"}
+        for _ in range(2):
+            st, _ = self.raw_post("/v1/link", body, hdrs)
+            self.assertNotEqual(st, 429)
+        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        self.addCleanup(conn.close)
+        conn.request("POST", "/v1/link", body=body, headers=hdrs)
+        resp = conn.getresponse()
+        self.assertEqual(resp.status, 429)
+        self.assertTrue(resp.getheader("Retry-After"))
+        self.assertIn("rate limit", json.loads(resp.read())["error"])
+
+    def test_health_is_never_rate_limited(self):
+        for _ in range(30):
+            conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+            conn.request("GET", "/health")
+            self.assertEqual(conn.getresponse().status, 200)
+            conn.close()
+
+
+class TestRateLimitHardening(unittest.TestCase):
+    """The four codex [P2] findings, each pinned by a test."""
+
+    def test_capacity_fails_closed_instead_of_resetting_a_live_window(self):
+        """Evicting a live window would let an attacker who can mint many
+        distinct keys clear their OWN limit on demand. New keys are refused
+        instead while every window is still live."""
+        now = [1000.0]
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (1, 3600)},
+                                      max_keys=5, clock=lambda: now[0])
+        for i in range(5):
+            self.assertTrue(rl.check("/v1/link", f"10.0.0.{i}")[0])
+        victim_hits = list(rl._hits["/v1/link"]["10.0.0.0"])
+        for i in range(100, 140):                 # identity flood
+            allowed, retry = rl.check("/v1/link", f"10.0.1.{i}")
+            self.assertFalse(allowed)
+            self.assertGreater(retry, 0)
+        # the earlier client's window survived the flood untouched
+        self.assertEqual(rl._hits["/v1/link"]["10.0.0.0"], victim_hits)
+        self.assertLessEqual(len(rl._hits["/v1/link"]), 5)
+
+    def test_capacity_recovers_once_windows_expire(self):
+        now = [1000.0]
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (1, 60)},
+                                      max_keys=3, clock=lambda: now[0])
+        for i in range(3):
+            rl.check("/v1/link", f"10.0.0.{i}")
+        self.assertFalse(rl.check("/v1/link", "10.0.9.9")[0])
+        now[0] += 61
+        self.assertTrue(rl.check("/v1/link", "10.0.9.9")[0])
+
+    def test_hits_stay_chronological_under_thread_contention(self):
+        """retry_after uses hits[0] and reclamation uses v[-1]; both assume
+        the list is ordered. Reading the clock outside the lock let racing
+        threads append out of order."""
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (500, 3600)})
+        ts = [threading.Thread(target=lambda: rl.check("/v1/link", "7.7.7.7"))
+              for _ in range(200)]
+        for t in ts:
+            t.start()
+        for t in ts:
+            t.join()
+        hits = rl._hits["/v1/link"]["7.7.7.7"]
+        self.assertEqual(len(hits), 200)
+        self.assertEqual(hits, sorted(hits), "timestamps out of order")
+
+
+class TestRateLimitConfigTypes(ServiceTestBase):
+    def _svc(self, rl):
+        cfg = dict(self.svc.cfg)
+        cfg["rate_limit"] = rl
+        return granthi_link.LinkService(cfg)
+
+    def test_non_boolean_enabled_refuses_startup(self):
+        """`"enabled": null` or `0` must not quietly mean unlimited."""
+        for bad in (None, 0, "", "false", "no", []):
+            with self.subTest(enabled=bad):
+                with self.assertRaises(SystemExit):
+                    self._svc({"enabled": bad})
+
+    def test_string_false_does_not_enable_forwarded_trust(self):
+        """Every non-empty string is truthy -- "false" used to mean True."""
+        with self.assertRaises(SystemExit):
+            self._svc({"trust_forwarded_for": "false",
+                       "trusted_proxies": ["10.0.0.0/8"]})
+
+    def test_rate_limit_must_be_an_object(self):
+        for bad in ("yes", 5, ["/v1/link"]):
+            with self.subTest(rl=bad):
+                with self.assertRaises(SystemExit):
+                    self._svc(bad)
+
+    def test_null_rate_limit_means_defaults_not_disabled(self):
+        svc = self._svc(None)
+        self.assertIsNotNone(svc.limiter)
+
+    def test_forwarded_trust_without_trusted_proxies_refuses_startup(self):
+        """Trusting the header from ANY peer lets callers choose their own
+        rate-limit key -- that must not be reachable by omission."""
+        with self.assertRaises(SystemExit):
+            self._svc({"trust_forwarded_for": True})
+        with self.assertRaises(SystemExit):
+            self._svc({"trust_forwarded_for": True, "trusted_proxies": []})
+
+    def test_forwarded_trust_with_proxies_is_accepted(self):
+        svc = self._svc({"trust_forwarded_for": True,
+                         "trusted_proxies": ["10.0.0.0/8", "127.0.0.1/32"]})
+        self.assertTrue(svc.trust_forwarded_for)
+        self.assertEqual(len(svc.trusted_proxies), 2)
+
+
+class TestRateLimitRound2(unittest.TestCase):
+    """The three findings from the codex re-review."""
+
+    def test_flooding_one_route_cannot_lock_out_another(self):
+        """A shared key table turned the fail-closed capacity guard into a
+        cross-route DoS: cheap /v1/repos keys could exhaust it and shut new
+        /v1/link clients out. Budgets are per route."""
+        now = [1000.0]
+        rl = granthi_link.RateLimiter(
+            rules={"/v1/link": (5, 3600), "/v1/repos": (60, 3600)},
+            max_keys=20, clock=lambda: now[0])
+        for i in range(300):                       # flood the cheap route
+            rl.check("/v1/repos", f"10.1.{i // 256}.{i % 256}")
+        allowed, _ = rl.check("/v1/link", "203.0.113.7")   # never-seen client
+        self.assertTrue(allowed, "/v1/link locked out by a /v1/repos flood")
+
+    def test_reclaim_uses_the_right_window_per_route(self):
+        now = [1000.0]
+        rl = granthi_link.RateLimiter(
+            rules={"/v1/link": (1, 3600), "/v1/repos": (1, 10)},
+            max_keys=2, clock=lambda: now[0])
+        rl.check("/v1/repos", "a")
+        rl.check("/v1/repos", "b")
+        self.assertFalse(rl.check("/v1/repos", "c")[0])
+        now[0] += 11                    # past the /v1/repos window only
+        self.assertTrue(rl.check("/v1/repos", "c")[0])
+
+
+class TestRateLimitConfigRound2(ServiceTestBase):
+    def _svc(self, rl):
+        cfg = dict(self.svc.cfg)
+        cfg["rate_limit"] = rl
+        return granthi_link.LinkService(cfg)
+
+    def test_booleans_are_rejected_in_rule_slots(self):
+        """bool subclasses int, so isinstance let [5, True] through as a
+        1-SECOND window -- 5/hour silently became ~5/sec."""
+        for bad in ([5, True], [True, 3600], [False, 3600], [5, False]):
+            with self.subTest(rule=bad):
+                with self.assertRaises(SystemExit):
+                    self._svc({"rules": {"/v1/link": bad}})
+
+    def test_valid_int_rule_still_accepted(self):
+        svc = self._svc({"rules": {"/v1/link": [5, 3600]}})
+        self.assertEqual(svc.limiter.rules["/v1/link"], (5, 3600))
+
+    def test_trusted_proxies_as_bare_string_refuses_startup(self):
+        """A string would be iterated character by character, each char
+        treated as a network."""
+        with self.assertRaises(SystemExit):
+            self._svc({"trust_forwarded_for": True,
+                       "trusted_proxies": "10.0.0.0/8"})
+
+    def test_wildcard_trusted_proxy_refuses_startup(self):
+        """0.0.0.0/0 or ::/0 restores 'trust XFF from anyone' -- the exact
+        hole trusted_proxies exists to close."""
+        for wild in ("0.0.0.0/0", "::/0"):
+            with self.subTest(cidr=wild):
+                with self.assertRaises(SystemExit):
+                    self._svc({"trust_forwarded_for": True,
+                               "trusted_proxies": [wild]})
+
+    def test_non_string_proxy_entry_refuses_startup(self):
+        with self.assertRaises(SystemExit):
+            self._svc({"trust_forwarded_for": True,
+                       "trusted_proxies": [10, "10.0.0.0/8"]})
+
+    def test_host_address_without_prefix_is_accepted(self):
+        svc = self._svc({"trust_forwarded_for": True,
+                         "trusted_proxies": ["127.0.0.1", "10.0.0.0/8"]})
+        self.assertEqual(len(svc.trusted_proxies), 2)