test: preserve empty credential helper reset

Hit live today. An operator moved an email onto a different forge account; the
next sign-in tried to create a user with that address, Gitea answered
422 "e-mail already in use", and the generic path turned it into a bare 502.
The person approved a device code and got a number that reads like an outage,
when the real answer was "that address already belongs to somebody here".

create_user now distinguishes that case, and the binding rules translate it to
409 with a message naming the address, the login it tried to create, and the
two ways out: have an operator bind your identity to the existing account, or
use a different address.

211 tests (was 209): the stub now enforces unique emails like real Gitea, one
test asserts the 409 names the address and login and never says 502, and one
asserts an ordinary create is unaffected.

README.md

@@ -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
+(see "Invite-only story"). You do **not** need to be on any private network:
-provisioning service is not publicly exposed yet.
+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)
@@ -211,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
+* `list [pattern]` — every repo the linked token can see, with the local
-  is already synced to. Reads `GET /api/v1/user/repos` on the forge
+  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
@@ -225,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
+* `add <folder> [--name N] [--private|--public] [--mode M] [--force]` —
-  needed, creates the cloud repo via `/v1/repos`, adds remote `granthi`,
+  `git init -b main` if needed, creates the cloud repo via `/v1/repos`, adds
-  initial commit + push. The token is delivered by a **git credential
+  remote `granthi`, initial commit + push. The token is delivered by a **git
-  helper** (the client's hidden `git-credential` subcommand reading the 0600
+  credential helper** (the client's hidden `git-credential` subcommand
-  config) — never embedded in the remote URL (estate rule).
+  reading the 0600 config) — never embedded in the remote URL (estate rule).
-* `watch [--interval 30] [--once]` — per folder: autocommit
+  Pushes the folder's **current** branch, not a hardcoded `main`: an existing
-  (`sync: <ISO ts>`) → fetch → ff-pull if remote strictly ahead → push if
+  repo may sit on `master` or a feature branch, and publishing that work
-  local strictly ahead. **DIVERGED → log + record + SKIP. Never force, never
+  under the wrong name is not a cosmetic error.
-  merge** — the same policy as the mesh. SIGTERM-clean.
+  Two guards, because `add -A` takes whatever it is given: a starter
-* `status` — table of linked folders, last sync, divergence flags.
+  `.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
@@ -267,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
+1. ~~**Expose :3042** behind cloudflared~~ **DONE 2026-08-23** —
-   link.granthi.shre.ai) — today it is tailnet-only by design.
+   `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

SETUP.md

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

server/granthi_link.py

@@ -64,6 +64,7 @@ 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
@@ -73,18 +74,37 @@ 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)}
+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
 
 
@@ -275,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.
 
@@ -322,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)
@@ -337,6 +527,7 @@ 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
@@ -497,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(
+        suffix = LOGIN_SAFE.sub("-", str(device_id or ""))[:12]
-            safe_dev, datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"))
+        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={
@@ -545,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)
@@ -570,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:
@@ -596,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)")
@@ -619,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,
+        # A device id the client generated once and keeps. Clients older than
-                     "token": gitea_token, "token_name": token_name}
+        # 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:
@@ -719,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)

tests/test_server.py

@@ -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,6 +564,475 @@ 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()