diff --git a/DESIGN.md b/DESIGN.md
index 661c739..f55bac9 100644
--- a/DESIGN.md
+++ b/DESIGN.md
@@ -352,9 +352,27 @@ The script reads GitHub; the model does not.
 
 Allowlist file `$AGENT_HOME/watch.json` key `assigned_repos` (non-empty list of `Owner/repo` strings). Missing or empty is an error; there is no default list.
 
-The first successful scan records `assigned_watch_since` and the assigned `session_id` and dispatches nothing. Later scans consider assignments whose latest matching `assigned` event is at or after that cursor, skipping `assigned_at` values already stored. Changing `session_id` after that pin is an error. The scan uses this device’s paired GitHub login; a missing pair or a `gh api user` mismatch is an error. The queue head is the already-knocked inflight item if any, then remaining items oldest first.
+The first successful scan records `assigned_watch_since` and the assigned `session_id` and dispatches nothing. Later scans consider assignments whose latest matching `assigned` event is at or after that cursor, skipping ones not newer than the stored `(assigned_at, event_id)` marker — same-second events only pass when the candidate's `event_id` is known and either the stored marker has none or the candidate's is higher, so a genuinely later same-second assignment discovered in a later scan is not silently dropped. Changing `session_id` after that pin is an error. The scan uses this device’s paired GitHub login; a missing pair or a `gh api user` mismatch is an error. The queue head is the already-knocked inflight item if any, then remaining items ordered by `(assigned_at, event_id)` — oldest first, and by `event_id` ascending among any that share a timestamp — so two rows a same-second reassignment left pending process in their real GitHub order.
 
-The writer is this device. All assignments share **one** runner session (`watch.json` `session_id`, default `assigned`, characters `A-Za-z0-9_-` only). That auto-created session attaches `spine`, `review-loop`, and `pr-review`; an existing row under the same id must already be `kind=runner`. Other sessions still attach skills themselves. There is one tmux/Grok terminal, not one per issue. Working files go to `$AGENT_HOME/sessions/<id>` or `$AGENT_SESSION_ROOT/<id>`. New `issue.assigned` rows enqueue on that session (`payload`: repo, number, url, title, body, assigned_at, assignee, mandate). The insert does not notify the knock daemon. The script pushes own events, writes `MANDATE.md` / `QUEUE.md` (no issue body), starts Grok only if that session is not already attached, then knocks at most the head of the queue (`da ist Post id <uuid>`). A knock of `issue.assigned` rewrites those files immediately before send. Further knocks stay queued until the **supervise script** records `issue.assigned.ack` with `payload.assigned_id`. The model must not insert that ack. The scan watermark `assigned_watch_since` is the scan clock, not the last seen GitHub event time — that is the no-backfill rule.
+The writer is this device. All assignments share **one** runner session (`watch.json` `session_id`, default `assigned`, characters `A-Za-z0-9_-` only). That auto-created session attaches `spine`, `review-loop`, and `pr-review`; an existing row under the same id must already be `kind=runner`. Other sessions still attach skills themselves. There is one tmux/Grok terminal, not one per issue. Working files go to `$AGENT_HOME/sessions/<id>` or `$AGENT_SESSION_ROOT/<id>`. New `issue.assigned` rows enqueue on that session (`payload`: repo, number, url, title, body, assigned_at, assigned_by, event_id (scan-only, see below), assignee, mandate). The insert does not notify the knock daemon. The script pushes own events, writes `MANDATE.md` / `QUEUE.md` (no issue body), starts Grok only if that session is not already attached, then knocks at most the head of the queue (`da ist Post id <uuid>`). A knock of `issue.assigned` rewrites those files immediately before send. Further knocks stay queued until the **supervise script** records `issue.assigned.ack` with `payload.assigned_id`. The model must not insert that ack. The scan watermark `assigned_watch_since` is the scan clock, not the last seen GitHub event time — that is the no-backfill rule.
+
+Optional `$AGENT_HOME/policy.json` uses the same fail-closed `admits()` gate as job admission. Before `supervise` commissions a queue head — including when the session's pane is already up — or `dispatch_assigned` starts/kicks a session, the script consults that file (when present): `private` comes from a live `gh api repos/<repo> --jq .private` lookup, or defaults to private when no runner is available. A denial leaves the queue head's own workspace files, wake claim, and session state untouched, so the next tick/poll re-evaluates it identically — the `sync()` housekeeping that `dispatch_assigned` runs before consulting the policy is unconditional and unrelated to this outcome; idle-clock bookkeeping (`_mark_working`), by contrast, is skipped on a denial along with everything else that commissioning would have done; `supervise` reports `supervise denied assigned=<id>`, and `agent watch assigned`'s own polling loop reports the same outcome as `assigned denied <id>`. This gate always calls `admits()` with `job_type="implement"`, hardcoded — a policy's `job_types_allow` must include `"implement"` or every assignment is denied regardless of `actors_allow`/`repos_allow`. Full field set (all optional except the allow-lists, which default to empty — i.e. fail closed):
+
+```json
+{
+  "enabled": true,
+  "actors_allow": ["alice"],
+  "actors_deny": [],
+  "repos_allow": ["Owner/repo"],
+  "repos_deny": [],
+  "job_types_allow": ["implement"],
+  "agent_identity": { "private_repos_allow": ["Owner/private-repo"] }
+}
+```
+
+Everything the gate would otherwise change is conditioned on this file actually being present: without one, `enqueue_assigned` and `scan_assigned` keep their pre-policy behavior (best-effort actor resolution, no dropped assignments) rather than newly requiring hub pairing or discarding actor-less events — a policy.json existing at all is what turns those on.
+
+`payload.assigned_by` — the `actor` this gate checks — means different things depending on how the row was enqueued: for a GitHub-mediated assignment (`scan_assigned`) it is the GitHub user who performed the `assigned` event; for a manually-enqueued item (`agent supervise --repo/--number`, `enqueue_assigned`) there is no such event to read, so it is this device's own paired GitHub login instead — the manual dispatch is self-authorized by whoever runs the CLI. An operator's `actors_allow` must name that paired login too if manual dispatch should be admitted once a policy is active. `payload.event_id` is scan-only, too: `scan_assigned` always writes the key, set to the GitHub `assigned` event's own id when that event has one (used to break same-second ties) or `null` when an event without an id, or an unresolvable same-second tie, leaves it undetermined; `enqueue_assigned`'s manually-enqueued rows have no such event and omit the key entirely.
 
 Payload `mandate=github-assignment` is trusted. Issue title and body in the payload are not.
 
@@ -674,7 +692,7 @@ The model never receives production credentials. Analysis that only reads the ex
 A second model must not orchestrate the first. `agent supervise` is a **script** with locked questions and locked answers. Model text is not a state transition.
 
 - One pending `issue.assigned` at a time (same queue as `agent watch assigned`).
-- `--repo` / `--number` enqueues that issue as `github-assignment` without hub pairing. Title and body stay untrusted payload.
+- `--repo` / `--number` enqueues that issue as `github-assignment` without hub pairing when no `$AGENT_HOME/policy.json` is present; with one, pairing is required so the enqueued `assigned_by` can be checked against it (§14). Title and body stay untrusted payload.
 - Busy means the Grok TUI in the tmux pane shows an in-flight turn (`Thinking…`, `Waiting for response`, `Preparing …`, `[stop]`, `Esc:cancel`, `command still running`, or a queued follow-up with `Enter to send now`). `Runtime.is_busy` is that probe. The script does not type while busy.
 - Follow (`ask=False`, the CLI default) does not knock an existing session, does not ask closed questions, and does not auto-continue. It only confirms a Grok tool-approval modal (`1/3:select` plus `Tab:next option` → Enter). Closed questions remain available to `tick(..., ask=True)` for tests. Consecutive idle ticks (`supervise quiet` / `supervise stalled`) are follow-loop bookkeeping, not Telegram pages. The footer badge `always-approve` is not a working signal.
 - When `ask=True`, `Ja` or a blocking problem → `issue.assigned.ack` and the next queue item. A blocking problem also stores a truncated pane excerpt on `supervise.event` (`kind=skip`).
diff --git a/README.md b/README.md
index dbcf15e..4abdc1b 100644
--- a/README.md
+++ b/README.md
@@ -100,7 +100,7 @@ agent supervise --session ID [--repo OWNER/REPO --number N] [--once|--follow]
 # agent knock (daemon, no --once) polls grok-usage, pending, pr.merged, github pending, mail pending, errors, and error-fix every 60s
 ```
 
-`agent supervise` posts a short status line to Telegram when both `TELEGRAM_BOT_TOKEN` and `TELEGRAM_CHAT_ID` are set in the environment. The follow CLI does not ask closed questions. Working vs not working for paging is whether the Grok tmux session exists: it posts `not working` only when that session is gone, not when the prompt is idle between turns. The TUI working probe (`Thinking…`, `Waiting for response`, `Preparing …`, `[stop]`, `Esc:cancel`, `command still running`, queued `Enter to send now`) is for the follow loop, not for Telegram. A send failure is printed to stderr and does not stop the loop. Credentials stay out of git.
+`agent supervise --repo/--number` enqueues that issue without hub pairing when no `$AGENT_HOME/policy.json` is present; with one, pairing is required, and a denied dispatch prints `supervise denied assigned=<id>` and leaves the queue head untouched (`agent watch assigned` reports the same outcome as `assigned denied <id>`) — see DESIGN.md for the policy format. `agent supervise` posts a short status line to Telegram when both `TELEGRAM_BOT_TOKEN` and `TELEGRAM_CHAT_ID` are set in the environment. The follow CLI does not ask closed questions. Working vs not working for paging is whether the Grok tmux session exists: it posts `not working` only when that session is gone, not when the prompt is idle between turns. The TUI working probe (`Thinking…`, `Waiting for response`, `Preparing …`, `[stop]`, `Esc:cancel`, `command still running`, queued `Enter to send now`) is for the follow loop, not for Telegram. A send failure is printed to stderr and does not stop the loop. Credentials stay out of git.
 
 The error-fix executor find-or-creates the implement task and isolated worktree; `agent github pending` still opens draft pull requests.
 
@@ -112,7 +112,7 @@ The error-fix executor find-or-creates the implement task and isolated worktree;
 { "assigned_repos": ["Owner/repo"], "session_id": "assigned" }
 ```
 
-Missing or empty `assigned_repos` is an error. `session_id` is optional, defaults to `assigned`, and may contain only `A-Za-z0-9_-`. A session already present under that id must be `kind=runner`. The auto-created runner session attaches `spine`, `review-loop`, and `pr-review` (those skills stay opt-in for every other session). The working directory is `$AGENT_HOME/sessions/<session_id>` unless `AGENT_SESSION_ROOT` is set. The first successful scan records the `assigned_watch_since` watermark and the assigned session id, and creates no activities. Changing `session_id` after that pin is an error. The scan uses the paired GitHub login; a missing pair or a `gh api user` mismatch is an error. Later scans enqueue `issue.assigned` on **that one** runner session, push to the hub, write `MANDATE.md` / `QUEUE.md`, and start Grok only if that session is not already attached. The insert does not notify the knock daemon. There is one terminal; further assignments wait in the knock queue until the supervise script records `issue.assigned.ack` with `payload.assigned_id` set to that activity id. The follow CLI does not ack from pane text. `MANDATE.md` lists session and activity ids. `QUEUE.md` lists ids and urls. Neither file contains issue bodies. Use `--follow` for a 30s loop, or cron for one-shot runs.
+Missing or empty `assigned_repos` is an error. `session_id` is optional, defaults to `assigned`, and may contain only `A-Za-z0-9_-`. A session already present under that id must be `kind=runner`. The auto-created runner session attaches `spine`, `review-loop`, and `pr-review` (those skills stay opt-in for every other session). The working directory is `$AGENT_HOME/sessions/<session_id>` unless `AGENT_SESSION_ROOT` is set. The first successful scan records the `assigned_watch_since` watermark and the assigned session id, and creates no activities. Changing `session_id` after that pin is an error. The scan uses the paired GitHub login; a missing pair or a `gh api user` mismatch is an error. Later scans enqueue `issue.assigned` on **that one** runner session, push to the hub, write `MANDATE.md` / `QUEUE.md`, and start Grok only if that session is not already attached. The insert does not notify the knock daemon. There is one terminal; further assignments wait in the knock queue until the supervise script records `issue.assigned.ack` with `payload.assigned_id` set to that activity id. The follow CLI does not ack from pane text. `MANDATE.md` lists session and activity ids. `QUEUE.md` lists ids and urls. Neither file contains issue bodies. Use `--follow` for a 30s loop, or cron for one-shot runs. An optional `$AGENT_HOME/policy.json` can gate which assignments this command will actually dispatch — a denial prints `assigned denied <id>` and leaves the queue head untouched; see DESIGN.md for the policy format.
 
 ### Session terminal control
 
diff --git a/src/agent_cli/main.py b/src/agent_cli/main.py
index f85449b..08207fd 100644
--- a/src/agent_cli/main.py
+++ b/src/agent_cli/main.py
@@ -3126,7 +3126,7 @@ def cmd_watch(args: list[str]) -> None:
                 if pending:
                     head_id = pending[0].get("id")
                     if isinstance(head_id, str) and head_id:
-                        dispatch_assigned(
+                        dispatched = dispatch_assigned(
                             store,
                             head_id,
                             sync=lambda: _sync_once(store),
@@ -3143,7 +3143,10 @@ def cmd_watch(args: list[str]) -> None:
                             knock=lambda activity_id: deliver(store, Runtime(), activity_id),
                             workspace_root=workspace_root,
                             pane_up=lambda session_id: Runtime().exists(session_id),
+                            runner=run_argv,
                         )
+                        if dispatched == "denied":
+                            print(f"assigned denied {head_id}")
                 for activity_id in created:
                     print(f"issue.assigned {activity_id}")
                 if not created:
@@ -3261,6 +3264,7 @@ def start(session_id: str, cwd: Path) -> None:
                 knock=lambda activity_id: deliver(store, runtime, activity_id),
                 pane=pane,
                 working=working,
+                runner=run_argv,
             )
             print(line)
             try:
diff --git a/src/agent_cli/supervise.py b/src/agent_cli/supervise.py
index 4bf0c18..7197c2e 100644
--- a/src/agent_cli/supervise.py
+++ b/src/agent_cli/supervise.py
@@ -23,9 +23,12 @@
 from .watch import (
     _ensure_assigned_session,
     _issue_number,
+    _paired_login,
+    _policy_admits,
     assigned_workspace_root,
     dispatch_assigned,
     pending_assigned,
+    policy_present,
 )
 
 ANSWER_YES = "Ja"
@@ -215,6 +218,19 @@ def enqueue_assigned(
     if existing is not None:
         return existing
     now = utcnow()
+    # Pairing is only load-bearing once a policy is active to check the
+    # actor against — without one, requiring it would break manual enqueue
+    # for operators who never opted into policy.json (DESIGN.md: enqueues
+    # "without hub pairing"). Still resolve it best-effort either way; only
+    # a policy in play makes a broken pairing fatal.
+    if policy_present(store.home):
+        assigned_by = _paired_login(store, runner)
+    else:
+        assigned_by = ""
+        try:
+            assigned_by = _paired_login(store, runner)
+        except StoreError:
+            pass
     _ensure_assigned_session(store, session_id, now)
     url = f"https://github.com/{repo}/issues/{number}"
     title = ""
@@ -259,6 +275,7 @@ def enqueue_assigned(
                 "title": title,
                 "body": body,
                 "assigned_at": now,
+                "assigned_by": assigned_by,
                 "assignee": assignee,
                 "mandate": "github-assignment",
             },
@@ -315,6 +332,7 @@ def tick(
     ask: bool = False,
     pane: str | None = None,
     working: bool | None = None,
+    runner: Callable[[list[str]], Completed] | None = None,
 ) -> str:
     if SESSION_RE.match(session_id) is None:
         raise StoreError("session id may contain only A-Za-z0-9_-")
@@ -343,6 +361,8 @@ def tick(
             last_answer = last_payload.get("answer")
     pane_missing = not runtime.exists(session_id)
     if last_kind is None and not pane_missing:
+        if not _policy_admits(store, head, runner):
+            return f"supervise denied assigned={assigned_id}"
         _log(
             store,
             session_id,
@@ -365,7 +385,10 @@ def tick(
             knock=knock,
             workspace_root=root,
             pane_up=lambda sid: runtime.exists(sid),
+            runner=runner,
         )
+        if dispatched == "denied":
+            return f"supervise denied assigned={assigned_id}"
     else:
         dispatched = "held"
     if last_kind is None:
diff --git a/src/agent_cli/watch.py b/src/agent_cli/watch.py
index 99fd06c..180c0c2 100644
--- a/src/agent_cli/watch.py
+++ b/src/agent_cli/watch.py
@@ -6,12 +6,15 @@
 import os
 import re
 import socket
+import stat
 import uuid
 from collections.abc import Callable
 from datetime import datetime
 from pathlib import Path
 from typing import Any
 
+from .ingest import _repo_is_private
+from .jobs import Verdict, admits
 from .runtime import Completed
 from .store import Store, StoreError
 
@@ -220,6 +223,47 @@ def load_watch_config(home: Path) -> tuple[list[str], str]:
     return list(repos), session_id
 
 
+def policy_present(home: Path) -> bool:
+    """Whether `home / "policy.json"` exists as something readable as a
+    policy. A directory, broken symlink, or other non-regular entry at that
+    path is a misconfiguration, not "no policy" — callers must not treat it
+    as the backward-compatible absent case. Uses os.lstat/os.stat directly
+    (rather than pathlib's is_file/exists/is_symlink, which swallow any
+    OSError — not just ENOENT — and would make a real stat failure such as
+    EACCES or ESTALE indistinguishable from "no policy")."""
+    path = home / "policy.json"
+    try:
+        lst = os.lstat(path)
+    except FileNotFoundError:
+        return False
+    except OSError as exc:
+        raise StoreError(f"{path} could not be checked: {exc}") from exc
+    if stat.S_ISLNK(lst.st_mode):
+        try:
+            st = os.stat(path)
+        except FileNotFoundError:
+            raise StoreError(f"{path} exists but is not a regular file") from None
+        except OSError as exc:
+            raise StoreError(f"{path} could not be checked: {exc}") from exc
+    else:
+        st = lst
+    if stat.S_ISREG(st.st_mode):
+        return True
+    raise StoreError(f"{path} exists but is not a regular file")
+
+
+def load_policy(home: Path) -> Any:
+    """The parsed `home / "policy.json"` object, or None if the file does not
+    exist or its JSON value is itself `null`."""
+    if not policy_present(home):
+        return None
+    path = home / "policy.json"
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        raise StoreError(f"{path} is invalid JSON") from exc
+
+
 def assigned_session_id(home: Path) -> str:
     _repos, sid = load_watch_config(home)
     return sid
@@ -242,9 +286,15 @@ def _paired_login(store: Store, runner: Callable[[list[str]], Completed]) -> str
     return paired.lower()
 
 
-def _latest_assigned_at(store: Store, repo: str, number: int) -> datetime | None:
+def _latest_assigned_marker(
+    store: Store, repo: str, number: int
+) -> tuple[datetime, int | None] | None:
+    """The (assigned_at, event_id) of the most recently recorded assignment
+    activity for repo/number, or None if none stored. Among rows tied on
+    assigned_at, the highest present event_id wins; None if none of the
+    tied rows have one."""
     repo_key = repo.lower()
-    latest: datetime | None = None
+    latest: tuple[datetime, int | None] | None = None
     for row in store.rows("activity"):
         if row.get("type") != "issue.assigned":
             continue
@@ -271,11 +321,37 @@ def _latest_assigned_at(store: Store, repo: str, number: int) -> datetime | None
             event_dt = _parse_gh_time(raw_at)
         except ValueError:
             continue
-        if latest is None or event_dt > latest:
-            latest = event_dt
+        raw_eid = payload.get("event_id")
+        event_id = raw_eid if isinstance(raw_eid, int) and not isinstance(raw_eid, bool) else None
+        if latest is None or event_dt > latest[0]:
+            latest = (event_dt, event_id)
+        elif event_dt == latest[0]:
+            if event_id is not None and (latest[1] is None or event_id > latest[1]):
+                latest = (event_dt, event_id)
     return latest
 
 
+def _assignment_is_newer(
+    dt: datetime, event_id: int | None, marker: tuple[datetime, int | None] | None
+) -> bool:
+    """Whether (dt, event_id) is provably newer than `marker`. A same-
+    timestamp tie needs the candidate's own id to be known: an unresolvable
+    candidate never wins (fail closed), but an unresolvable STORED marker
+    (a legacy row predating this field) must not permanently block every
+    future candidate at that timestamp — so a resolvable candidate beats a
+    marker with no id."""
+    if marker is None:
+        return True
+    prev_dt, prev_id = marker
+    if dt > prev_dt:
+        return True
+    if dt < prev_dt:
+        return False
+    if event_id is None:
+        return False
+    return prev_id is None or event_id > prev_id
+
+
 def _ensure_assigned_session(store: Store, sid: str, now: str) -> None:
     existing = store.row("session", sid)
     if existing is None:
@@ -384,6 +460,8 @@ def scan_assigned(
                 continue
             newest_at: str | None = None
             newest_dt: datetime | None = None
+            newest_by = ""
+            newest_id: int | None = None
             for event in events:
                 if not isinstance(event, dict):
                     continue
@@ -404,13 +482,37 @@ def scan_assigned(
                     continue
                 if event_dt < cursor_dt:
                     continue
+                actor = event.get("actor")
+                assigned_by = ""
+                if isinstance(actor, dict):
+                    actor_login = actor.get("login")
+                    if isinstance(actor_login, str):
+                        assigned_by = actor_login
+                raw_id = event.get("id")
+                event_id = raw_id if isinstance(raw_id, int) and not isinstance(raw_id, bool) else None
                 if newest_dt is None or event_dt > newest_dt:
                     newest_dt = event_dt
                     newest_at = created_at
+                    newest_by = assigned_by
+                    newest_id = event_id
+                elif event_dt == newest_dt:
+                    if event_id is not None and newest_id is not None and event_id > newest_id:
+                        newest_by = assigned_by
+                        newest_id = event_id
+                    elif event_id is None or newest_id is None:
+                        newest_by = ""
+                        newest_id = None
             if newest_at is None or newest_dt is None:
                 continue
-            previous_at = _latest_assigned_at(store, repo, number)
-            if previous_at is not None and newest_dt <= previous_at:
+            # An unresolvable actor only jams the queue once a policy is
+            # active to reject it (admits() denies on a missing actor and
+            # nothing acks a denial). Without a policy, drop nothing that
+            # GitHub reported as assigned — that would be a needless
+            # behavior change for operators who never opted into policy.json.
+            if newest_by == "" and policy_present(store.home):
+                continue
+            previous = _latest_assigned_marker(store, repo, number)
+            if not _assignment_is_newer(newest_dt, newest_id, previous):
                 continue
             title = issue.get("title")
             body = issue.get("body")
@@ -423,6 +525,8 @@ def scan_assigned(
                     "title": title if isinstance(title, str) else "",
                     "body": body if isinstance(body, str) else "",
                     "assigned_at": newest_at,
+                    "assigned_by": newest_by,
+                    "event_id": newest_id,
                 }
             )
     found.sort(key=lambda item: (str(item["assigned_at"]), str(item["repo"]).lower(), int(item["number"])))
@@ -434,12 +538,13 @@ def scan_assigned(
         repo = str(item["repo"])
         number = int(item["number"])
         assigned_at = str(item["assigned_at"])
+        event_id = item["event_id"]
         try:
             assigned_dt = _parse_gh_time(assigned_at)
         except ValueError:
             continue
-        previous_at = _latest_assigned_at(store, repo, number)
-        if previous_at is not None and assigned_dt <= previous_at:
+        previous = _latest_assigned_marker(store, repo, number)
+        if not _assignment_is_newer(assigned_dt, event_id, previous):
             continue
         activity_id = str(uuid.uuid4())
         lock_key = f"assigned:{repo.lower()}:{number}:{assigned_at}"
@@ -458,15 +563,16 @@ def scan_assigned(
                     "title": item["title"],
                     "body": item["body"],
                     "assigned_at": assigned_at,
+                    "assigned_by": item["assigned_by"],
+                    "event_id": event_id,
                     "assignee": login,
                     "mandate": "github-assignment",
                 },
                 "execution_status": "done",
             },
             lock_key=lock_key,
-            skip=lambda r=repo, n=number, dt=assigned_dt: (
-                _latest_assigned_at(store, r, n) is not None
-                and _latest_assigned_at(store, r, n) >= dt
+            skip=lambda r=repo, n=number, dt=assigned_dt, eid=event_id: (
+                not _assignment_is_newer(dt, eid, _latest_assigned_marker(store, r, n))
             ),
         )
         if event is not None:
@@ -510,7 +616,7 @@ def acked_assigned_ids(store: Store, session_id: str) -> set[str]:
 
 def pending_assigned(store: Store, session_id: str) -> list[dict[str, Any]]:
     acked = acked_assigned_ids(store, session_id)
-    ranked: list[tuple[str, str, dict[str, Any]]] = []
+    ranked: list[tuple[str, int, str, dict[str, Any]]] = []
     for row in store.rows("activity"):
         if row.get("type") != "issue.assigned":
             continue
@@ -523,13 +629,22 @@ def pending_assigned(store: Store, session_id: str) -> list[dict[str, Any]]:
             continue
         payload = row.get("payload")
         assigned_at = ""
-        if isinstance(payload, dict) and isinstance(payload.get("assigned_at"), str):
-            assigned_at = payload["assigned_at"]
-        ranked.append((assigned_at, aid, row))
+        event_id = -1
+        if isinstance(payload, dict):
+            if isinstance(payload.get("assigned_at"), str):
+                assigned_at = payload["assigned_at"]
+            raw_id = payload.get("event_id")
+            if isinstance(raw_id, int) and not isinstance(raw_id, bool):
+                event_id = raw_id
+        # Same-timestamp rows can coexist (a later scan may add one for a
+        # genuinely later same-second GitHub event, see _assignment_is_newer)
+        # — order those by event_id, not by the random activity uuid, so the
+        # chronologically later one is processed after the earlier one.
+        ranked.append((assigned_at, event_id, aid, row))
     ranked.sort()
     inflight: list[dict[str, Any]] = []
     rest: list[dict[str, Any]] = []
-    for _assigned_at, aid, row in ranked:
+    for _assigned_at, _event_id, aid, row in ranked:
         if store.wake_delivered(aid):
             inflight.append(row)
         else:
@@ -578,6 +693,30 @@ def _write_assigned_queue_files(
     (cwd / "QUEUE.md").write_text("\n".join(queue), encoding="utf-8")
 
 
+def _policy_admits(
+    store: Store,
+    head: dict[str, Any],
+    runner: Callable[[list[str]], Completed] | None,
+) -> bool:
+    """Whether `head` (an issue.assigned activity row) is admitted by the
+    policy at `store.home / "policy.json"`. No policy file present admits
+    unconditionally — this is what keeps the gate backward-compatible."""
+    if not policy_present(store.home):
+        return True
+    policy = load_policy(store.home)
+    payload = head.get("payload")
+    payload = payload if isinstance(payload, dict) else {}
+    repo = payload.get("repo")
+    repo = repo if isinstance(repo, str) else ""
+    assigned_by = payload.get("assigned_by")
+    assigned_by = assigned_by if isinstance(assigned_by, str) else ""
+    private = True if runner is None else _repo_is_private(repo, runner)
+    verdict: Verdict = admits(
+        policy, actor=assigned_by, repo=repo, job_type="implement", private=private
+    )
+    return verdict.admitted
+
+
 def dispatch_assigned(
     store: Store,
     activity_id: str,
@@ -587,6 +726,7 @@ def dispatch_assigned(
     knock: Callable[[str], Any],
     workspace_root: Path,
     pane_up: Callable[[str], bool] | None = None,
+    runner: Callable[[list[str]], Completed] | None = None,
 ) -> str:
     activity = store.row("activity", activity_id)
     if activity is None:
@@ -615,6 +755,12 @@ def dispatch_assigned(
     head_id = head.get("id")
     if not isinstance(head_id, str) or head_id == "":
         raise StoreError(f"session {sid} queue head is missing an id")
+    # Denial must not create this head's workspace files, claim its wake entry, or
+    # start a session, so the same head re-evaluates identically next tick — this
+    # does not cover the sync() call above, which is local to dispatch_assigned
+    # and always runs before the policy check regardless of the denial outcome.
+    if not _policy_admits(store, head, runner):
+        return "denied"
     cwd = workspace_root / sid
     cwd.mkdir(parents=True, exist_ok=True)
     _write_assigned_queue_files(cwd, sid, head, pending)
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 2614c6a..6963738 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -2077,6 +2077,28 @@ def test_watch_assigned_requires_watch_json(tmp_path: Path) -> None:
         run(tmp_path, ["watch", "assigned"])
 
 
+def test_watch_assigned_prints_denied_on_policy_rejection(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    run(tmp_path, ["init"])
+    (tmp_path / "watch.json").write_text(
+        json.dumps({"assigned_repos": ["Owner/repo"]}), encoding="utf-8"
+    )
+    monkeypatch.setattr("agent_cli.main.scan_assigned", lambda store, runner, now: ([], 0))
+    monkeypatch.setattr(
+        "agent_cli.main.pending_assigned",
+        lambda store, sid: [{"id": "asg-1"}],
+    )
+    monkeypatch.setattr(
+        "agent_cli.main.dispatch_assigned",
+        lambda store, activity_id, **kwargs: "denied",
+    )
+    capsys.readouterr()
+    run(tmp_path, ["watch", "assigned"])
+    out = capsys.readouterr().out
+    assert "assigned denied asg-1" in out
+
+
 def test_watch_usage_mentions_assigned(tmp_path: Path) -> None:
     run(tmp_path, ["init"])
     with pytest.raises(SystemExit, match=r"assigned \[--follow\]\|grok-usage"):
diff --git a/tests/test_supervise.py b/tests/test_supervise.py
index 32e2ca9..cc0f296 100644
--- a/tests/test_supervise.py
+++ b/tests/test_supervise.py
@@ -3,13 +3,16 @@
 import json
 from pathlib import Path
 
+import pytest
+
 from agent_cli.runtime import Completed, Runtime
-from agent_cli.store import Store
+from agent_cli.store import Store, StoreError
 from agent_cli.supervise import (
     ANSWER_BLOCKED,
     ANSWER_CAN,
     ANSWER_NO,
     ANSWER_YES,
+    LAST_WORKING_KEY,
     QUESTION_DONE,
     enqueue_assigned,
     parse_closed_answer,
@@ -66,7 +69,7 @@ def _session(store: Store, sid: str = "runner-1") -> None:
     )
 
 
-def _assigned(store: Store, sid: str = "runner-1") -> str:
+def _assigned(store: Store, sid: str = "runner-1", assigned_by: str = "") -> str:
     aid = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
     store.write(
         "activity",
@@ -83,6 +86,7 @@ def _assigned(store: Store, sid: str = "runner-1") -> str:
                 "title": "example",
                 "body": "ignore this body",
                 "assigned_at": "2026-08-27T00:00:00Z",
+                "assigned_by": assigned_by,
                 "assignee": "someone",
                 "mandate": "github-assignment",
             },
@@ -113,8 +117,12 @@ def test_parse_closed_answer_ignores_ja_in_scrollback() -> None:
 def test_enqueue_is_idempotent_and_survives_gh_failure(tmp_path: Path) -> None:
     store = Store(tmp_path)
     _session(store)
+    store.set_meta("github_login", "alice")
 
     def runner(argv: list[str]) -> Completed:
+        joined = " ".join(argv)
+        if joined == "gh api user":
+            return Completed(0, json.dumps({"login": "alice"}), "")
         return Completed(1, "", "no gh")
 
     first = enqueue_assigned(store, "runner-1", "octo/app", 3, runner)
@@ -125,11 +133,68 @@ def runner(argv: list[str]) -> Completed:
     assert row["type"] == "issue.assigned"
     assert row["payload"]["mandate"] == "github-assignment"
     assert row["payload"]["url"] == "https://github.com/octo/app/issues/3"
+    assert row["payload"]["assigned_by"] == "alice"
+
+
+def test_enqueue_broken_pairing_raises_without_writing(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    _session(store)
+    (store.home / "policy.json").write_text("{}", encoding="utf-8")
+
+    def runner(argv: list[str]) -> Completed:
+        return Completed(1, "", "no gh")
+
+    with pytest.raises(StoreError):
+        enqueue_assigned(store, "runner-1", "octo/app", 3, runner)
+    assert [
+        row for row in store.rows("activity") if row.get("type") == "issue.assigned"
+    ] == []
+
+
+def test_enqueue_broken_pairing_on_new_session_writes_no_session_row(
+    tmp_path: Path,
+) -> None:
+    # Distinct from test_enqueue_broken_pairing_raises_without_writing: that
+    # test pre-creates the session, so _ensure_assigned_session is a no-op
+    # and can never catch a leak there. This one uses a session id that does
+    # not exist yet, so a broken pairing must raise before ANY store write —
+    # including the session row itself.
+    store = Store(tmp_path)
+    (store.home / "policy.json").write_text("{}", encoding="utf-8")
+
+    def runner(argv: list[str]) -> Completed:
+        return Completed(1, "", "no gh")
+
+    with pytest.raises(StoreError):
+        enqueue_assigned(store, "brand-new", "octo/app", 3, runner)
+    assert store.row("session", "brand-new") is None
+    assert [
+        row for row in store.rows("activity") if row.get("type") == "issue.assigned"
+    ] == []
+
+
+def test_enqueue_broken_pairing_without_a_policy_degrades_instead_of_raising(
+    tmp_path: Path,
+) -> None:
+    # Mirrors test_enqueue_broken_pairing_raises_without_writing but with no
+    # policy.json: without an active policy, a broken pairing must not block
+    # manual enqueue at all (DESIGN.md: enqueues "without hub pairing").
+    store = Store(tmp_path)
+    _session(store)
+
+    def runner(argv: list[str]) -> Completed:
+        return Completed(1, "", "no gh")
+
+    aid = enqueue_assigned(store, "runner-1", "octo/app", 3, runner)
+    row = store.row("activity", aid)
+    assert row is not None
+    assert row["payload"]["assigned_by"] == ""
 
 
 def test_enqueue_uses_gh_json(tmp_path: Path) -> None:
     store = Store(tmp_path)
     _session(store)
+    store.set_meta("github_login", "octocat")
     body = {
         "title": "t",
         "body": "b",
@@ -138,6 +203,9 @@ def test_enqueue_uses_gh_json(tmp_path: Path) -> None:
     }
 
     def runner(argv: list[str]) -> Completed:
+        joined = " ".join(argv)
+        if joined == "gh api user":
+            return Completed(0, json.dumps({"login": "octocat"}), "")
         assert argv[:3] == ["gh", "api", "repos/octo/app/issues/3"]
         return Completed(0, json.dumps(body), "")
 
@@ -146,6 +214,114 @@ def runner(argv: list[str]) -> Completed:
     assert row is not None
     assert row["payload"]["assignee"] == "octocat"
     assert row["payload"]["title"] == "t"
+    assert row["payload"]["assigned_by"] == "octocat"
+
+
+def test_tick_denies_and_does_not_mutate_when_policy_rejects(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    _session(store)
+    assigned = _assigned(store, assigned_by="mallory")
+    (store.home / "policy.json").write_text(
+        json.dumps(
+            {
+                "actors_allow": ["alice"],
+                "repos_allow": ["octo/app"],
+                "job_types_allow": ["implement"],
+            }
+        ),
+        encoding="utf-8",
+    )
+    def runner(argv: list[str]) -> Completed:
+        if ".private" in " ".join(argv):
+            return Completed(0, "false", "")
+        raise AssertionError(argv)
+
+    rt = FakeRuntime(exists=False, pane="")
+    line = tick(
+        store,
+        rt,
+        "runner-1",
+        start=lambda sid, cwd: None,
+        knock=lambda aid: "sent",
+        runner=runner,
+    )
+    assert line == f"supervise denied assigned={assigned}"
+    events = [r for r in store.rows("activity") if r.get("type") == "supervise.event"]
+    assert events == []
+    assert store.sync_get(LAST_WORKING_KEY) is None
+
+
+def test_tick_denies_pane_up_and_does_not_mutate_when_policy_rejects(
+    tmp_path: Path,
+) -> None:
+    store = Store(tmp_path)
+    _session(store)
+    assigned = _assigned(store, assigned_by="mallory")
+    (store.home / "policy.json").write_text(
+        json.dumps(
+            {
+                "actors_allow": ["alice"],
+                "repos_allow": ["octo/app"],
+                "job_types_allow": ["implement"],
+            }
+        ),
+        encoding="utf-8",
+    )
+    def runner(argv: list[str]) -> Completed:
+        if ".private" in " ".join(argv):
+            return Completed(0, "false", "")
+        raise AssertionError(argv)
+
+    rt = FakeRuntime(exists=True, pane="")
+    line = tick(
+        store,
+        rt,
+        "runner-1",
+        start=lambda sid, cwd: None,
+        knock=lambda aid: "sent",
+        runner=runner,
+    )
+    assert line == f"supervise denied assigned={assigned}"
+    events = [r for r in store.rows("activity") if r.get("type") == "supervise.event"]
+    assert events == []
+    assert store.sync_get(LAST_WORKING_KEY) is None
+
+
+def test_tick_commissions_when_policy_admits_pane_up(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    _session(store)
+    assigned = _assigned(store, assigned_by="alice")
+    (store.home / "policy.json").write_text(
+        json.dumps(
+            {
+                "actors_allow": ["alice"],
+                "repos_allow": ["octo/app"],
+                "job_types_allow": ["implement"],
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    def runner(argv: list[str]) -> Completed:
+        joined = " ".join(argv)
+        if ".private" in joined:
+            return Completed(0, "false", "")
+        raise AssertionError(argv)
+
+    rt = FakeRuntime(exists=True, pane="")
+    line = tick(
+        store,
+        rt,
+        "runner-1",
+        start=lambda sid, cwd: None,
+        knock=lambda aid: "sent",
+        runner=runner,
+    )
+    assert line == f"supervise commission assigned={assigned} dispatch=held"
+    events = [r for r in store.rows("activity") if r.get("type") == "supervise.event"]
+    assert len(events) == 1
+    assert events[0]["payload"]["kind"] == "commission"
+    assert store.sync_get(LAST_WORKING_KEY) is not None
 
 
 def test_tick_commissions_then_asks_then_acks_yes(tmp_path: Path) -> None:
diff --git a/tests/test_watch.py b/tests/test_watch.py
index ce0bb5e..a43cdf9 100644
--- a/tests/test_watch.py
+++ b/tests/test_watch.py
@@ -5,13 +5,16 @@
 
 import pytest
 
+from agent_cli import watch
 from agent_cli.runtime import Completed
 from agent_cli.store import Store, StoreError
 from agent_cli.watch import (
     ISSUE_LIST_LIMIT,
     dispatch_assigned,
+    load_policy,
     load_watch_config,
     pending_assigned,
+    policy_present,
     scan_assigned,
     scan_merged,
 )
@@ -288,6 +291,7 @@ def runner(argv: list[str]) -> Completed:
                             "event": "assigned",
                             "created_at": "2026-01-01T00:00:00Z",
                             "assignee": {"login": "alice"},
+                            "actor": {"login": "bob"},
                         }
                     ]
                 ),
@@ -303,6 +307,7 @@ def runner(argv: list[str]) -> Completed:
     assert row["type"] == "issue.assigned"
     assert row["payload"]["number"] == 8
     assert row["payload"]["mandate"] == "github-assignment"
+    assert row["payload"]["assigned_by"] == "bob"
     assert row["session_id"] == "assigned"
     session = store.row("session", row["session_id"])
     assert session is not None
@@ -313,6 +318,416 @@ def runner(argv: list[str]) -> Completed:
     assert again == []
 
 
+def test_scan_assigned_missing_actor_skips_without_persisting(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    store.set_meta("github_login", "alice")
+    _write_assigned_repos(tmp_path)
+    _write_admit_policy(tmp_path)
+    store.sync_set("assigned_watch_since", "2020-01-01T00:00:00Z")
+
+    def runner(argv: list[str]) -> Completed:
+        if argv[:3] == ["gh", "api", "user"]:
+            return Completed(0, json.dumps({"login": "alice"}), "")
+        if argv[:3] == ["gh", "issue", "list"]:
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "number": 8,
+                            "title": "Fix it",
+                            "url": "https://github.com/Owner/repo/issues/8",
+                            "body": "",
+                        }
+                    ]
+                ),
+                "",
+            )
+        if argv[:2] == ["gh", "api"] and any("events" in part for part in argv):
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "event": "assigned",
+                            "created_at": "2026-01-01T00:00:00Z",
+                            "assignee": {"login": "alice"},
+                        }
+                    ]
+                ),
+                "",
+            )
+        raise AssertionError(f"unexpected argv: {argv}")
+
+    created, skipped = scan_assigned(store, runner, now="2026-08-23T12:00:00Z")
+    assert created == []
+    assert skipped == 0
+
+
+def test_scan_assigned_missing_actor_still_persists_without_a_policy(
+    tmp_path: Path,
+) -> None:
+    # Same fixture as test_scan_assigned_missing_actor_skips_without_persisting,
+    # minus the policy.json: without an active policy there is nothing to jam,
+    # so this must keep the pre-policy behavior of enqueueing what GitHub
+    # reported as assigned, even with a blank assigned_by.
+    store = Store(tmp_path)
+    store.set_meta("github_login", "alice")
+    _write_assigned_repos(tmp_path)
+    store.sync_set("assigned_watch_since", "2020-01-01T00:00:00Z")
+
+    def runner(argv: list[str]) -> Completed:
+        if argv[:3] == ["gh", "api", "user"]:
+            return Completed(0, json.dumps({"login": "alice"}), "")
+        if argv[:3] == ["gh", "issue", "list"]:
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "number": 8,
+                            "title": "Fix it",
+                            "url": "https://github.com/Owner/repo/issues/8",
+                            "body": "",
+                        }
+                    ]
+                ),
+                "",
+            )
+        if argv[:2] == ["gh", "api"] and any("events" in part for part in argv):
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "event": "assigned",
+                            "created_at": "2026-01-01T00:00:00Z",
+                            "assignee": {"login": "alice"},
+                        }
+                    ]
+                ),
+                "",
+            )
+        raise AssertionError(f"unexpected argv: {argv}")
+
+    created, skipped = scan_assigned(store, runner, now="2026-08-23T12:00:00Z")
+    assert skipped == 0
+    assert len(created) == 1
+    row = store.row("activity", created[0])
+    assert row is not None
+    assert row["payload"]["assigned_by"] == ""
+
+
+def test_scan_assigned_same_second_uses_higher_event_id(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    store.set_meta("github_login", "alice")
+    _write_assigned_repos(tmp_path)
+    store.sync_set("assigned_watch_since", "2020-01-01T00:00:00Z")
+
+    def runner(argv: list[str]) -> Completed:
+        if argv[:3] == ["gh", "api", "user"]:
+            return Completed(0, json.dumps({"login": "alice"}), "")
+        if argv[:3] == ["gh", "issue", "list"]:
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "number": 8,
+                            "title": "Fix it",
+                            "url": "https://github.com/Owner/repo/issues/8",
+                            "body": "SECRET_BODY_DO_NOT_COPY",
+                        }
+                    ]
+                ),
+                "",
+            )
+        if argv[:2] == ["gh", "api"] and any("events" in part for part in argv):
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "id": 100,
+                            "event": "assigned",
+                            "created_at": "2026-01-01T00:00:00Z",
+                            "assignee": {"login": "alice"},
+                            "actor": {"login": "first"},
+                        },
+                        {
+                            "id": 200,
+                            "event": "assigned",
+                            "created_at": "2026-01-01T00:00:00Z",
+                            "assignee": {"login": "alice"},
+                            "actor": {"login": "later"},
+                        },
+                    ]
+                ),
+                "",
+            )
+        raise AssertionError(f"unexpected argv: {argv}")
+
+    created, skipped = scan_assigned(store, runner, now="2026-08-23T12:00:00Z")
+    assert skipped == 0
+    assert len(created) == 1
+    row = store.row("activity", created[0])
+    assert row is not None
+    assert row["payload"]["assigned_by"] == "later"
+    assert row["payload"]["event_id"] == 200
+
+
+def test_scan_assigned_same_second_unresolvable_tie_skips_without_persisting(
+    tmp_path: Path,
+) -> None:
+    store = Store(tmp_path)
+    store.set_meta("github_login", "alice")
+    _write_assigned_repos(tmp_path)
+    _write_admit_policy(tmp_path)
+    store.sync_set("assigned_watch_since", "2020-01-01T00:00:00Z")
+
+    def runner(argv: list[str]) -> Completed:
+        if argv[:3] == ["gh", "api", "user"]:
+            return Completed(0, json.dumps({"login": "alice"}), "")
+        if argv[:3] == ["gh", "issue", "list"]:
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "number": 8,
+                            "title": "Fix it",
+                            "url": "https://github.com/Owner/repo/issues/8",
+                            "body": "SECRET_BODY_DO_NOT_COPY",
+                        }
+                    ]
+                ),
+                "",
+            )
+        if argv[:2] == ["gh", "api"] and any("events" in part for part in argv):
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "id": 100,
+                            "event": "assigned",
+                            "created_at": "2026-01-01T00:00:00Z",
+                            "assignee": {"login": "alice"},
+                            "actor": {"login": "first"},
+                        },
+                        {
+                            "event": "assigned",
+                            "created_at": "2026-01-01T00:00:00Z",
+                            "assignee": {"login": "alice"},
+                            "actor": {"login": "middle"},
+                        },
+                        {
+                            "id": 200,
+                            "event": "assigned",
+                            "created_at": "2026-01-01T00:00:00Z",
+                            "assignee": {"login": "alice"},
+                            "actor": {"login": "later"},
+                        },
+                    ]
+                ),
+                "",
+            )
+        raise AssertionError(f"unexpected argv: {argv}")
+
+    created, skipped = scan_assigned(store, runner, now="2026-08-23T12:00:00Z")
+    assert created == []
+    assert skipped == 0
+
+
+def test_scan_assigned_resolvable_candidate_beats_a_blanked_stored_marker(
+    tmp_path: Path,
+) -> None:
+    store = Store(tmp_path)
+    store.set_meta("github_login", "alice")
+    _write_assigned_repos(tmp_path)
+    store.sync_set("assigned_watch_since", "2020-01-01T00:00:00Z")
+    # Legacy stored marker with no event_id (field added later): scan_assigned
+    # no longer manufactures these, but a resolvable candidate at the same
+    # timestamp must still beat a stored marker with no id.
+    store.write(
+        "session",
+        "insert",
+        "assigned",
+        {
+            "id": "assigned",
+            "kind": "runner",
+            "status": "active",
+            "started_at": "2026-01-01T00:00:00Z",
+            "last_seen_at": "2026-01-01T00:00:00Z",
+            "host": "test",
+        },
+    )
+    store.sync_set("assigned_session_id", "assigned")
+    store.write(
+        "activity",
+        "insert",
+        "legacy-blanked",
+        {
+            "id": "legacy-blanked",
+            "session_id": "assigned",
+            "type": "issue.assigned",
+            "payload": {
+                "repo": "Owner/repo",
+                "number": 8,
+                "url": "https://github.com/Owner/repo/issues/8",
+                "title": "Fix it",
+                "body": "SECRET_BODY_DO_NOT_COPY",
+                "assigned_at": "2026-01-01T00:00:00Z",
+                "assigned_by": "",
+                "event_id": None,
+                "mandate": "github-assignment",
+            },
+            "execution_status": "done",
+        },
+    )
+
+    def runner(argv: list[str]) -> Completed:
+        if argv[:3] == ["gh", "api", "user"]:
+            return Completed(0, json.dumps({"login": "alice"}), "")
+        if argv[:3] == ["gh", "issue", "list"]:
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "number": 8,
+                            "title": "Fix it",
+                            "url": "https://github.com/Owner/repo/issues/8",
+                            "body": "SECRET_BODY_DO_NOT_COPY",
+                        }
+                    ]
+                ),
+                "",
+            )
+        if argv[:2] == ["gh", "api"] and any("events" in part for part in argv):
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "id": 100,
+                            "event": "assigned",
+                            "created_at": "2026-01-01T00:00:00Z",
+                            "assignee": {"login": "alice"},
+                            "actor": {"login": "first"},
+                        },
+                        {
+                            "id": 300,
+                            "event": "assigned",
+                            "created_at": "2026-01-01T00:00:00Z",
+                            "assignee": {"login": "alice"},
+                            "actor": {"login": "resolved"},
+                        },
+                    ]
+                ),
+                "",
+            )
+        raise AssertionError(f"unexpected argv: {argv}")
+
+    created, skipped = scan_assigned(store, runner, now="2026-08-23T13:00:00Z")
+    assert skipped == 0
+    assert len(created) == 1
+    row = store.row("activity", created[0])
+    assert row is not None
+    assert row["id"] != "legacy-blanked"
+    assert row["payload"]["assigned_by"] == "resolved"
+    assert row["payload"]["event_id"] == 300
+
+
+def test_scan_assigned_same_second_higher_event_id_across_scans(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    store.set_meta("github_login", "alice")
+    _write_assigned_repos(tmp_path)
+    store.sync_set("assigned_watch_since", "2020-01-01T00:00:00Z")
+    events_calls = {"n": 0}
+
+    def runner(argv: list[str]) -> Completed:
+        if argv[:3] == ["gh", "api", "user"]:
+            return Completed(0, json.dumps({"login": "alice"}), "")
+        if argv[:3] == ["gh", "issue", "list"]:
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "number": 8,
+                            "title": "Fix it",
+                            "url": "https://github.com/Owner/repo/issues/8",
+                            "body": "SECRET_BODY_DO_NOT_COPY",
+                        }
+                    ]
+                ),
+                "",
+            )
+        if argv[:2] == ["gh", "api"] and any("events" in part for part in argv):
+            events_calls["n"] += 1
+            if events_calls["n"] == 1:
+                return Completed(
+                    0,
+                    json.dumps(
+                        [
+                            {
+                                "id": 100,
+                                "event": "assigned",
+                                "created_at": "2026-01-01T00:00:00Z",
+                                "assignee": {"login": "alice"},
+                                "actor": {"login": "A"},
+                            }
+                        ]
+                    ),
+                    "",
+                )
+            return Completed(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "id": 100,
+                            "event": "assigned",
+                            "created_at": "2026-01-01T00:00:00Z",
+                            "assignee": {"login": "alice"},
+                            "actor": {"login": "A"},
+                        },
+                        {
+                            "id": 200,
+                            "event": "assigned",
+                            "created_at": "2026-01-01T00:00:00Z",
+                            "assignee": {"login": "alice"},
+                            "actor": {"login": "B"},
+                        },
+                    ]
+                ),
+                "",
+            )
+        raise AssertionError(f"unexpected argv: {argv}")
+
+    created, skipped = scan_assigned(store, runner, now="2026-01-01T00:00:00Z")
+    assert skipped == 0
+    assert len(created) == 1
+    row = store.row("activity", created[0])
+    assert row is not None
+    assert row["payload"]["assigned_by"] == "A"
+    assert row["payload"]["event_id"] == 100
+
+    # `now` here must not advance the watch cursor past the events'
+    # `created_at` (both "2026-01-01T00:00:00Z"), or scan_assigned's
+    # no-backfill rule filters them out before the tie-break/marker
+    # comparison this test exercises ever runs.
+    created2, skipped2 = scan_assigned(store, runner, now="2026-08-23T13:00:00Z")
+    assert skipped2 == 0
+    assert len(created2) == 1
+    row2 = store.row("activity", created2[0])
+    assert row2 is not None
+    assert row2["id"] != created[0]
+    assert row2["payload"]["assigned_by"] == "B"
+    assert row2["payload"]["event_id"] == 200
+
+
 def test_scan_assigned_does_not_mutate_existing_runner_skills(tmp_path: Path) -> None:
     store = Store(tmp_path)
     store.set_meta("github_login", "alice")
@@ -357,6 +772,7 @@ def runner(argv: list[str]) -> Completed:
                             "event": "assigned",
                             "created_at": "2026-01-01T00:00:00Z",
                             "assignee": {"login": "alice"},
+                            "actor": {"login": "alice"},
                         }
                     ]
                 ),
@@ -405,6 +821,7 @@ def runner(argv: list[str]) -> Completed:
                             "event": "assigned",
                             "created_at": "2026-01-01T00:00:00Z",
                             "assignee": {"login": "alice"},
+                            "actor": {"login": "alice"},
                         }
                     ]
                 ),
@@ -515,7 +932,108 @@ def runner(argv: list[str]) -> Completed:
     assert store.sync_get("assigned_watch_since") is None
 
 
-def test_dispatch_assigned_sync_failure_does_not_start(tmp_path: Path) -> None:
+def test_dispatch_assigned_sync_failure_does_not_start(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    sid = "assigned"
+    store.write(
+        "session",
+        "insert",
+        sid,
+        {"id": sid, "kind": "runner", "status": "active"},
+    )
+    store.write(
+        "activity",
+        "insert",
+        "asg-1",
+        {
+            "id": "asg-1",
+            "session_id": sid,
+            "type": "issue.assigned",
+            "payload": {
+                "repo": "Owner/repo",
+                "number": 8,
+                "url": "https://github.com/Owner/repo/issues/8",
+                "title": "t",
+                "body": "SECRET_BODY_DO_NOT_COPY",
+                "mandate": "github-assignment",
+            },
+            "execution_status": "done",
+        },
+    )
+    start_log: list[tuple[str, Path]] = []
+    knock_log: list[str] = []
+
+    def sync() -> None:
+        raise StoreError("hub down")
+
+    with pytest.raises(StoreError, match="hub down"):
+        dispatch_assigned(
+            store,
+            "asg-1",
+            sync=sync,
+            start=lambda s, cwd: start_log.append((s, cwd)),
+            knock=lambda aid: knock_log.append(aid),
+            workspace_root=tmp_path / "sessions",
+        )
+    assert start_log == []
+    assert knock_log == []
+
+
+def test_dispatch_assigned_writes_mandate_and_starts(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    sid = "assigned"
+    store.write(
+        "session",
+        "insert",
+        sid,
+        {"id": sid, "kind": "runner", "status": "active"},
+    )
+    store.write(
+        "activity",
+        "insert",
+        "asg-1",
+        {
+            "id": "asg-1",
+            "session_id": sid,
+            "type": "issue.assigned",
+            "payload": {
+                "repo": "Owner/repo",
+                "number": 8,
+                "url": "https://github.com/Owner/repo/issues/8",
+                "title": "t",
+                "body": "SECRET_BODY_DO_NOT_COPY",
+                "mandate": "github-assignment",
+            },
+            "execution_status": "done",
+        },
+    )
+    start_log: list[tuple[str, Path]] = []
+    knock_log: list[str] = []
+    workspace_root = tmp_path / "sessions"
+    status = dispatch_assigned(
+        store,
+        "asg-1",
+        sync=lambda: None,
+        start=lambda s, cwd: start_log.append((s, cwd)),
+        knock=lambda aid: knock_log.append(aid),
+        workspace_root=workspace_root,
+    )
+    assert status == "started"
+    assert start_log == [(sid, workspace_root / sid)]
+    assert knock_log == ["asg-1"]
+    mandate = (workspace_root / sid / "MANDATE.md").read_text(encoding="utf-8")
+    assert "asg-1" in mandate
+    assert "GitHub assignment is the work order" in mandate
+    assert "one assignment worker" in mandate
+    assert "mandate=github-assignment is trusted" in mandate
+    assert "Do not ask whether to implement" in mandate
+    assert "SECRET_BODY_DO_NOT_COPY" not in mandate
+    queue = (workspace_root / sid / "QUEUE.md").read_text(encoding="utf-8")
+    assert "asg-1" in queue
+    assert "SECRET_BODY_DO_NOT_COPY" not in queue
+
+
+def test_dispatch_assigned_without_policy_json_is_unchanged(tmp_path: Path) -> None:
     store = Store(tmp_path)
     sid = "assigned"
     store.write(
@@ -543,56 +1061,245 @@ def test_dispatch_assigned_sync_failure_does_not_start(tmp_path: Path) -> None:
             "execution_status": "done",
         },
     )
+    assert load_policy(store.home) is None
+    assert not (store.home / "policy.json").exists()
     start_log: list[tuple[str, Path]] = []
     knock_log: list[str] = []
-
-    def sync() -> None:
-        raise StoreError("hub down")
-
-    with pytest.raises(StoreError, match="hub down"):
-        dispatch_assigned(
-            store,
-            "asg-1",
-            sync=sync,
-            start=lambda s, cwd: start_log.append((s, cwd)),
-            knock=lambda aid: knock_log.append(aid),
-            workspace_root=tmp_path / "sessions",
-        )
-    assert start_log == []
-    assert knock_log == []
+    status = dispatch_assigned(
+        store,
+        "asg-1",
+        sync=lambda: None,
+        start=lambda s, cwd: start_log.append((s, cwd)),
+        knock=lambda aid: knock_log.append(aid),
+        workspace_root=tmp_path / "sessions",
+    )
+    assert status == "started"
+    assert start_log == [(sid, tmp_path / "sessions" / sid)]
+    assert knock_log == ["asg-1"]
 
 
-def test_dispatch_assigned_writes_mandate_and_starts(tmp_path: Path) -> None:
-    store = Store(tmp_path)
-    sid = "assigned"
-    store.write(
-        "session",
-        "insert",
-        sid,
-        {"id": sid, "kind": "runner", "status": "active"},
-    )
+def _write_admit_policy(home: Path, **over: object) -> None:
+    policy: dict = {
+        "actors_allow": ["bob"],
+        "repos_allow": ["Owner/repo"],
+        "job_types_allow": ["implement"],
+    }
+    policy.update(over)
+    (home / "policy.json").write_text(json.dumps(policy), encoding="utf-8")
+
+
+def _insert_assigned_activity(
+    store: Store,
+    *,
+    sid: str = "assigned",
+    aid: str = "asg-1",
+    attached: bool = False,
+    assigned_by: str = "bob",
+    repo: str = "Owner/repo",
+) -> None:
+    session: dict = {"id": sid, "kind": "runner", "status": "active"}
+    if attached:
+        session["runtime"] = {"control": "attached"}
+    store.write("session", "insert", sid, session)
     store.write(
         "activity",
         "insert",
-        "asg-1",
+        aid,
         {
-            "id": "asg-1",
+            "id": aid,
             "session_id": sid,
             "type": "issue.assigned",
             "payload": {
-                "repo": "Owner/repo",
+                "repo": repo,
                 "number": 8,
-                "url": "https://github.com/Owner/repo/issues/8",
+                "url": f"https://github.com/{repo}/issues/8",
                 "title": "t",
                 "body": "SECRET_BODY_DO_NOT_COPY",
+                "assigned_by": assigned_by,
                 "mandate": "github-assignment",
             },
             "execution_status": "done",
         },
     )
+
+
+def test_dispatch_assigned_denies_when_policy_rejects_actor(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    _insert_assigned_activity(store)
+    _write_admit_policy(tmp_path, actors_allow=["alice"])
+    start_log: list[tuple[str, Path]] = []
+    knock_log: list[str] = []
+    workspace_root = tmp_path / "sessions"
+
+    def runner(argv: list[str]) -> Completed:
+        if ".private" in " ".join(argv):
+            return Completed(0, "false", "")
+        raise AssertionError(argv)
+
+    status = dispatch_assigned(
+        store,
+        "asg-1",
+        sync=lambda: None,
+        start=lambda s, cwd: start_log.append((s, cwd)),
+        knock=lambda aid: knock_log.append(aid),
+        workspace_root=workspace_root,
+        runner=runner,
+    )
+    assert status == "denied"
+    assert start_log == []
+    assert knock_log == []
+    assert not (workspace_root / "assigned" / "MANDATE.md").exists()
+    assert store.row("activity", "asg-1") is not None
+    assert not store.wake_delivered("asg-1")
+
+
+def test_dispatch_assigned_denies_when_job_types_allow_omits_implement(
+    tmp_path: Path,
+) -> None:
+    store = Store(tmp_path)
+    _insert_assigned_activity(store)
+    _write_admit_policy(tmp_path, job_types_allow=["pr-review"])
+    start_log: list[tuple[str, Path]] = []
+    knock_log: list[str] = []
+
+    def runner(argv: list[str]) -> Completed:
+        if ".private" in " ".join(argv):
+            return Completed(0, "false", "")
+        raise AssertionError(argv)
+
+    status = dispatch_assigned(
+        store,
+        "asg-1",
+        sync=lambda: None,
+        start=lambda s, cwd: start_log.append((s, cwd)),
+        knock=lambda aid: knock_log.append(aid),
+        workspace_root=tmp_path / "sessions",
+        runner=runner,
+    )
+    assert status == "denied"
+    assert start_log == []
+    assert knock_log == []
+
+
+def test_dispatch_assigned_denies_when_policy_json_is_null(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    _insert_assigned_activity(store)
+    (tmp_path / "policy.json").write_text("null", encoding="utf-8")
+    start_log: list[tuple[str, Path]] = []
+    knock_log: list[str] = []
+    workspace_root = tmp_path / "sessions"
+    status = dispatch_assigned(
+        store,
+        "asg-1",
+        sync=lambda: None,
+        start=lambda s, cwd: start_log.append((s, cwd)),
+        knock=lambda aid: knock_log.append(aid),
+        workspace_root=workspace_root,
+    )
+    assert status == "denied"
+    assert start_log == []
+    assert knock_log == []
+    assert not (workspace_root / "assigned" / "MANDATE.md").exists()
+    assert store.row("activity", "asg-1") is not None
+    assert not store.wake_delivered("asg-1")
+
+
+def test_load_policy_raises_on_invalid_json(tmp_path: Path) -> None:
+    (tmp_path / "policy.json").write_text("{not valid json", encoding="utf-8")
+    with pytest.raises(StoreError, match="invalid JSON"):
+        load_policy(tmp_path)
+
+
+def test_policy_present_false_when_absent(tmp_path: Path) -> None:
+    assert policy_present(tmp_path) is False
+
+
+def test_policy_present_raises_when_policy_json_is_a_directory(tmp_path: Path) -> None:
+    (tmp_path / "policy.json").mkdir()
+    with pytest.raises(StoreError, match="not a regular file"):
+        policy_present(tmp_path)
+
+
+def test_policy_present_raises_when_policy_json_is_a_broken_symlink(
+    tmp_path: Path,
+) -> None:
+    (tmp_path / "policy.json").symlink_to(tmp_path / "missing-target")
+    with pytest.raises(StoreError, match="not a regular file"):
+        policy_present(tmp_path)
+
+
+def test_policy_present_raises_instead_of_silently_absent_on_stat_error(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    (tmp_path / "policy.json").write_text("{}", encoding="utf-8")
+
+    def fake_lstat(path: object, *args: object, **kwargs: object) -> object:
+        raise PermissionError(13, "Permission denied")
+
+    monkeypatch.setattr(watch.os, "lstat", fake_lstat)
+    with pytest.raises(StoreError, match="could not be checked"):
+        policy_present(tmp_path)
+
+
+def test_dispatch_assigned_raises_when_policy_json_is_invalid(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    _insert_assigned_activity(store)
+    (tmp_path / "policy.json").write_text("{not valid json", encoding="utf-8")
+    with pytest.raises(StoreError, match="invalid JSON"):
+        dispatch_assigned(
+            store,
+            "asg-1",
+            sync=lambda: None,
+            start=lambda s, cwd: None,
+            knock=lambda aid: None,
+            workspace_root=tmp_path / "sessions",
+        )
+
+
+def test_dispatch_assigned_denies_when_policy_rejects_attached(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    _insert_assigned_activity(store, attached=True)
+    _write_admit_policy(tmp_path, actors_allow=["alice"])
+    start_log: list[tuple[str, Path]] = []
+    knock_log: list[str] = []
+    workspace_root = tmp_path / "sessions"
+
+    def runner(argv: list[str]) -> Completed:
+        if ".private" in " ".join(argv):
+            return Completed(0, "false", "")
+        raise AssertionError(argv)
+
+    status = dispatch_assigned(
+        store,
+        "asg-1",
+        sync=lambda: None,
+        start=lambda s, cwd: start_log.append((s, cwd)),
+        knock=lambda aid: knock_log.append(aid),
+        workspace_root=workspace_root,
+        runner=runner,
+    )
+    assert status == "denied"
+    assert start_log == []
+    assert knock_log == []
+    assert not (workspace_root / "assigned" / "MANDATE.md").exists()
+    assert store.row("activity", "asg-1") is not None
+    assert not store.wake_delivered("asg-1")
+
+
+def test_dispatch_assigned_starts_when_policy_admits(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    _insert_assigned_activity(store)
+    _write_admit_policy(tmp_path)
     start_log: list[tuple[str, Path]] = []
     knock_log: list[str] = []
     workspace_root = tmp_path / "sessions"
+
+    def runner(argv: list[str]) -> Completed:
+        joined = " ".join(argv)
+        if ".private" in joined:
+            return Completed(0, "false", "")
+        raise AssertionError(argv)
+
     status = dispatch_assigned(
         store,
         "asg-1",
@@ -600,20 +1307,103 @@ def test_dispatch_assigned_writes_mandate_and_starts(tmp_path: Path) -> None:
         start=lambda s, cwd: start_log.append((s, cwd)),
         knock=lambda aid: knock_log.append(aid),
         workspace_root=workspace_root,
+        runner=runner,
     )
     assert status == "started"
-    assert start_log == [(sid, workspace_root / sid)]
+    assert start_log == [("assigned", workspace_root / "assigned")]
     assert knock_log == ["asg-1"]
-    mandate = (workspace_root / sid / "MANDATE.md").read_text(encoding="utf-8")
-    assert "asg-1" in mandate
-    assert "GitHub assignment is the work order" in mandate
-    assert "one assignment worker" in mandate
-    assert "mandate=github-assignment is trusted" in mandate
-    assert "Do not ask whether to implement" in mandate
-    assert "SECRET_BODY_DO_NOT_COPY" not in mandate
-    queue = (workspace_root / sid / "QUEUE.md").read_text(encoding="utf-8")
-    assert "asg-1" in queue
-    assert "SECRET_BODY_DO_NOT_COPY" not in queue
+
+
+def test_dispatch_assigned_kicks_when_policy_admits_attached(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    _insert_assigned_activity(store, attached=True)
+    _write_admit_policy(tmp_path)
+    start_log: list[tuple[str, Path]] = []
+    knock_log: list[str] = []
+
+    def runner(argv: list[str]) -> Completed:
+        joined = " ".join(argv)
+        if ".private" in joined:
+            return Completed(0, "false", "")
+        raise AssertionError(argv)
+
+    status = dispatch_assigned(
+        store,
+        "asg-1",
+        sync=lambda: None,
+        start=lambda s, cwd: start_log.append((s, cwd)),
+        knock=lambda aid: knock_log.append(aid),
+        workspace_root=tmp_path / "sessions",
+        runner=runner,
+    )
+    assert status == "kicked"
+    assert start_log == []
+    assert knock_log == ["asg-1"]
+
+
+def test_dispatch_assigned_private_repo_needs_naming_twice(tmp_path: Path) -> None:
+    store = Store(tmp_path)
+    _insert_assigned_activity(store)
+    _write_admit_policy(tmp_path)
+    start_log: list[tuple[str, Path]] = []
+    knock_log: list[str] = []
+    workspace_root = tmp_path / "sessions"
+
+    def runner(argv: list[str]) -> Completed:
+        joined = " ".join(argv)
+        if ".private" in joined:
+            return Completed(0, "true", "")
+        raise AssertionError(argv)
+
+    denied = dispatch_assigned(
+        store,
+        "asg-1",
+        sync=lambda: None,
+        start=lambda s, cwd: start_log.append((s, cwd)),
+        knock=lambda aid: knock_log.append(aid),
+        workspace_root=workspace_root,
+        runner=runner,
+    )
+    assert denied == "denied"
+    assert start_log == []
+    assert knock_log == []
+    _write_admit_policy(
+        tmp_path,
+        agent_identity={"private_repos_allow": ["Owner/repo"]},
+    )
+    admitted = dispatch_assigned(
+        store,
+        "asg-1",
+        sync=lambda: None,
+        start=lambda s, cwd: start_log.append((s, cwd)),
+        knock=lambda aid: knock_log.append(aid),
+        workspace_root=workspace_root,
+        runner=runner,
+    )
+    assert admitted == "started"
+    assert start_log == [("assigned", workspace_root / "assigned")]
+    assert knock_log == ["asg-1"]
+
+
+def test_dispatch_assigned_denies_when_runner_missing_and_repo_not_private_allowed(
+    tmp_path: Path,
+) -> None:
+    store = Store(tmp_path)
+    _insert_assigned_activity(store)
+    _write_admit_policy(tmp_path)
+    start_log: list[tuple[str, Path]] = []
+    knock_log: list[str] = []
+    status = dispatch_assigned(
+        store,
+        "asg-1",
+        sync=lambda: None,
+        start=lambda s, cwd: start_log.append((s, cwd)),
+        knock=lambda aid: knock_log.append(aid),
+        workspace_root=tmp_path / "sessions",
+    )
+    assert status == "denied"
+    assert start_log == []
+    assert knock_log == []
 
 
 def test_dispatch_assigned_kicks_when_attached(tmp_path: Path) -> None:
@@ -702,6 +1492,7 @@ def runner(argv: list[str]) -> Completed:
                             "event": "assigned",
                             "created_at": "2026-02-01T00:00:00Z",
                             "assignee": {"login": "alice"},
+                            "actor": {"login": "alice"},
                         }
                     ]
                 ),
@@ -716,6 +1507,7 @@ def runner(argv: list[str]) -> Completed:
                             "event": "assigned",
                             "created_at": "2026-01-01T00:00:00Z",
                             "assignee": {"login": "alice"},
+                            "actor": {"login": "alice"},
                         }
                     ]
                 ),
@@ -845,6 +1637,62 @@ def test_pending_assigned_keeps_delivered_inflight_as_head(tmp_path: Path) -> No
     assert [row["id"] for row in pending] == ["asg-old"]
 
 
+def test_pending_assigned_orders_same_timestamp_rows_by_event_id(
+    tmp_path: Path,
+) -> None:
+    # Two rows can share an assigned_at when a later scan adds a genuinely
+    # later same-second event (see _assignment_is_newer) — order those by
+    # event_id, not by the random activity uuid. Ids are chosen so that
+    # sorting by id alone (the pre-fix behavior) would give the WRONG
+    # order, so this genuinely fails without the event_id-aware sort key.
+    store = Store(tmp_path)
+    store.write(
+        "session",
+        "insert",
+        "assigned",
+        {"id": "assigned", "kind": "runner", "status": "active"},
+    )
+    store.write(
+        "activity",
+        "insert",
+        "asg-aaa-higher-event-id",
+        {
+            "id": "asg-aaa-higher-event-id",
+            "session_id": "assigned",
+            "type": "issue.assigned",
+            "payload": {
+                "repo": "Owner/repo",
+                "number": 8,
+                "assigned_at": "2026-06-01T00:00:00Z",
+                "event_id": 200,
+            },
+            "execution_status": "done",
+        },
+    )
+    store.write(
+        "activity",
+        "insert",
+        "asg-zzz-lower-event-id",
+        {
+            "id": "asg-zzz-lower-event-id",
+            "session_id": "assigned",
+            "type": "issue.assigned",
+            "payload": {
+                "repo": "Owner/repo",
+                "number": 8,
+                "assigned_at": "2026-06-01T00:00:00Z",
+                "event_id": 100,
+            },
+            "execution_status": "done",
+        },
+    )
+    pending = pending_assigned(store, "assigned")
+    assert [row["id"] for row in pending] == [
+        "asg-zzz-lower-event-id",
+        "asg-aaa-higher-event-id",
+    ]
+
+
 def test_scan_assigned_reassignment_while_pending_enqueues(tmp_path: Path) -> None:
     store = Store(tmp_path)
     store.set_meta("github_login", "alice")
@@ -900,6 +1748,7 @@ def runner(argv: list[str]) -> Completed:
                             "event": "assigned",
                             "created_at": "2026-06-01T00:00:00Z",
                             "assignee": {"login": "alice"},
+                            "actor": {"login": "alice"},
                         }
                     ]
                 ),
@@ -979,6 +1828,7 @@ def runner(argv: list[str]) -> Completed:
                             "event": "assigned",
                             "created_at": "2026-06-01T00:00:00Z",
                             "assignee": {"login": "alice"},
+                            "actor": {"login": "alice"},
                         }
                     ]
                 ),
@@ -1122,6 +1972,7 @@ def runner(argv: list[str]) -> Completed:
                             "event": "assigned",
                             "created_at": "2026-01-01T00:00:00Z",
                             "assignee": {"login": "alice"},
+                            "actor": {"login": "alice"},
                         }
                     ]
                 ),