Harden ops run identity and leases

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a028de-e2c8-7732-8521-46a7fc5db82f

docs/adr/adr-005-ops-runs-vs-dev-work-records.md

@@ -96,13 +96,20 @@ When a definition emits a TaskSpec for internal fleet work:
 3. Do **not** open a Forgejo issue.  
 4. Do **not** create a workplan task file for that day’s fire.
 
-Claim API (sketch; implement in ACTIVITY-WP-0026):
+Claim API (implemented in ACTIVITY-WP-0026 and hardened in
+ACTIVITY-WP-0036):
 
 - `POST /ops-runs/claim` — lease next open run matching labels / worker id  
 - `POST /ops-runs/{id}/complete` — succeeded + completion metadata  
 - `POST /ops-runs/{id}/fail` — failed + retry policy  
 - `GET /ops-runs?state=open` — operator visibility  
 
+Worker mutation credentials are bound to one configured queue `worker_id`;
+the request body cannot assert a different claim owner. Heartbeat, completion,
+and failure lock the row and require its lease deadline to remain strictly in
+the future. Operator/SSO credentials provide visibility and explicit
+administration, but do not act as a normal worker identity.
+
 activity-core remains **when / what / where** only: it does **not** run
 domain LLM sessions or hold tenant git credentials.
 

docs/deploy-ops-run-queue-railiance.md

@@ -90,8 +90,9 @@ explicit is better for ops):
 | `OPS_RUN_LEASE_SECONDS` | `900` | Optional |
 | `OPS_RUN_MAX_ATTEMPTS` | `3` | Optional |
 | `OPS_RUN_SLA_HOURS` | `1` | Status stuck threshold |
+| `ACTIVITY_CORE_WORKER_ID` | `rein-aharness@railiance01` | Exact non-secret identity bound to the worker token |
 
-Optional worker auth (recommended before external claim):
+Worker auth (required before external claim):
 
 ```bash
 # Generate once; store in secret — do not commit
@@ -101,6 +102,10 @@ kubectl -n activity-core patch secret actcore-runtime-secret --type merge \
 # Record token in operator secret store (OpenBao / password manager), not chat.
 ```
 
+Apply the ConfigMap identity and Secret token in the same rollout. If the token
+is present without `ACTIVITY_CORE_WORKER_ID`, worker mutations fail with 503;
+if a request body names another identity, they fail with 403.
+
 Apply:
 
 ```bash
@@ -223,11 +228,11 @@ curl -sS "http://127.0.0.1:8010/ops/automations/status?since=today" \
 ### 6. Claim path smoke (manual, no harness yet)
 
 ```bash
-WORKER_TOKEN=…   # from secret if set; else local-dev open auth
+WORKER_TOKEN=…   # from secret
 curl -sS -X POST "http://127.0.0.1:8010/ops-runs/claim" \
   -H "Content-Type: application/json" \
   -H "X-Worker-Token: ${WORKER_TOKEN}" \
-  -d '{"worker_id":"smoke@railiance01","labels":["automated"],"limit":1,"lease_seconds":120}' \
+  -d '{"worker_id":"rein-aharness@railiance01","labels":["automated"],"limit":1,"lease_seconds":120}' \
   | python3 -m json.tool
 ```
 
@@ -239,7 +244,7 @@ RUN_ID=…   # from claim response
 curl -sS -X POST "http://127.0.0.1:8010/ops-runs/${RUN_ID}/fail" \
   -H "Content-Type: application/json" \
   -H "X-Worker-Token: ${WORKER_TOKEN}" \
-  -d '{"worker_id":"smoke@railiance01","error":"T07 smoke only","reopen":true}'
+  -d '{"worker_id":"rein-aharness@railiance01","error":"T07 smoke only","reopen":true}'
 ```
 
 - [ ] Claim returns the open run  

docs/ops-run-queue.md

@@ -59,7 +59,9 @@ workplan task file. Not an issue-core or Forgejo ticket.
 - `labels_mode`: `any` (default) — run must contain at least one listed label;
   `all` — run must contain every listed label; omit `labels` to claim any open run.
 - Claim uses `FOR UPDATE SKIP LOCKED` for concurrency safety.
-- Stale claims (`state=claimed` and `lease_until < now()`) are reopened before select.
+- Stale claims (`state=claimed` and `lease_until <= now()`) are reopened before select.
+- Heartbeat, complete, and fail lock the row and require an active lease
+  (`lease_until > now()`). An expired worker cannot revive or close its claim.
 
 ### Complete / fail body
 
@@ -150,9 +152,14 @@ model output are persisted or returned.
 ## Auth
 
 - **Worker:** `ACTIVITY_CORE_WORKER_TOKEN` via `X-Worker-Token` or
-  `Authorization: Bearer` (same value accepted on claim/complete/fail/heartbeat).  
+  `Authorization: Bearer`, bound to the exact non-secret
-- **Operator:** existing ops SSO / `ACTIVITY_CORE_OPERATOR_TOKEN` for list/status.  
+  `ACTIVITY_CORE_WORKER_ID`. The body `worker_id` is a compatibility field and
-- **Local dev:** `ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS=1` when no tokens set.
+  must match that authenticated identity on claim/complete/fail/heartbeat.
+- **Operator:** existing ops SSO / `ACTIVITY_CORE_OPERATOR_TOKEN` for
+  list/status and explicit lease expiry; operator credentials are not accepted
+  as worker mutation identities.
+- **Local dev:** `ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS=1` is required when
+  no tokens are set. There is no implicit open mode.
 
 ## Env
 
@@ -162,6 +169,7 @@ model output are persisted or returned.
 | `OPS_RUN_LEASE_SECONDS` | `900` | Default claim lease |
 | `OPS_RUN_MAX_ATTEMPTS` | `3` | Fail permanently after N claims |
 | `ACTIVITY_CORE_WORKER_TOKEN` | unset | Harness claim credential |
+| `ACTIVITY_CORE_WORKER_ID` | unset | Exact queue identity bound to the worker credential; required when the token is set |
 
 ## Consumer (rein-aharness)
 

docs/runbook.md

@@ -456,7 +456,7 @@ curl -sS -X POST "http://localhost:8010/ops-runs/claim" \
 
 # Reopen stale leases
 curl -sS -X POST "http://localhost:8010/ops-runs/expire-leases" \
-  -H "X-Worker-Token: $ACTIVITY_CORE_WORKER_TOKEN"
+  -H "X-Operator-Token: $ACTIVITY_CORE_OPERATOR_TOKEN"
 ```
 
 | Env | Default | Meaning |
@@ -466,6 +466,12 @@ curl -sS -X POST "http://localhost:8010/ops-runs/expire-leases" \
 | `OPS_RUN_MAX_ATTEMPTS` | `3` | Fail permanently after N claims |
 | `OPS_RUN_SLA_HOURS` | `1` | Stuck threshold in status |
 | `ACTIVITY_CORE_WORKER_TOKEN` | unset | Harness claim auth |
+| `ACTIVITY_CORE_WORKER_ID` | unset | Exact worker identity bound to claim auth |
+
+Worker mutations reject a body `worker_id` that differs from
+`ACTIVITY_CORE_WORKER_ID`, and heartbeat/complete/fail reject a missing or
+expired lease. Set the token and identity together before starting the claim
+consumer.
 
 Glas-backed workers may return the Glas 1.0 `GatewayResult` on complete or
 fail. Activity Core persists only `result.execution_evidence` plus the existing

docs/task-emission-consumer-contract.md

@@ -22,6 +22,11 @@ issue-core and not a Forgejo ticket.
 | Complete | `POST /ops-runs/{id}/complete` | `{ worker_id, result }` |
 | Fail | `POST /ops-runs/{id}/fail` | `{ worker_id, error, reopen? }` |
 
+The worker token is bound by Activity Core to one configured `worker_id`.
+Every worker mutation must name that exact identity and must still hold a lease
+whose deadline is strictly in the future. Operator/SSO authentication does not
+substitute for worker authentication on these calls.
+
 Full field list, auth, and env: **`docs/ops-run-queue.md`**.  
 Consumer implementation (rein-aharness): **REIN-A-0002**.
 

k8s/railiance/20-runtime.yaml

@@ -24,7 +24,9 @@ data:
   OPS_RUN_LEASE_SECONDS: "900"
   OPS_RUN_MAX_ATTEMPTS: "3"
   OPS_RUN_SLA_HOURS: "1"
-  # ACTIVITY_CORE_WORKER_TOKEN lives in actcore-runtime-secret (optional until REIN-A-0002)
+  # Non-secret identity bound to ACTIVITY_CORE_WORKER_TOKEN at the API boundary.
+  ACTIVITY_CORE_WORKER_ID: rein-aharness@railiance01
+  # ACTIVITY_CORE_WORKER_TOKEN lives in actcore-runtime-secret.
   ACTIVITY_DEFINITION_DIRS: /etc/activity-core/external-definitions
   CUSTODIAN_REPO_ROOT: /var/custodian
   ACTIVITY_CORE_ROOT: /etc/activity-core

k8s/railiance/README.md

@@ -41,8 +41,10 @@ host-network bridge and workstation tunnel are not part of this deployment.
 `OPS_RUN_QUEUE_ENABLED=true`. After image + migrate job (alembic **0007**),
 workers insert claimable `ops_runs` on emit. Full railiance checklist:
 `docs/deploy-ops-run-queue-railiance.md`. Keep host timers until REIN-A-0002.
-Optional `ACTIVITY_CORE_WORKER_TOKEN` in `actcore-runtime-secret` for harness
+`ACTIVITY_CORE_WORKER_TOKEN` in `actcore-runtime-secret` authenticates the
-claim auth.
+harness claim client. The non-secret `ACTIVITY_CORE_WORKER_ID` in the runtime
+ConfigMap binds that credential to `rein-aharness@railiance01`; deploy both
+settings together.
 
 | ExternalSecret | OpenBao path | Secret key |
 | --- | --- | --- |

k8s/railiance/bootstrap-secrets.sh

@@ -43,6 +43,6 @@ fi
 # actcore-issue-core-runtime (k8s/railiance/15-externalsecret-issue-core.yaml).
 # Apply that manifest after ClusterSecretStore openbao-activity-core is Ready.
 #
-# Optional (ACTIVITY-WP-0026 / REIN-A-0002): patch ACTIVITY_CORE_WORKER_TOKEN and
+# Patch ACTIVITY_CORE_WORKER_TOKEN and ACTIVITY_CORE_OPERATOR_TOKEN into
-# ACTIVITY_CORE_OPERATOR_TOKEN into actcore-runtime-secret — see
+# actcore-runtime-secret before enabling their API paths — see
 # docs/deploy-ops-run-queue-railiance.md. Not auto-generated here.

src/activity_core/ops_run_queue.py

@@ -36,6 +36,20 @@ def default_lease_seconds() -> int:
         return 900
 
 
+def _utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _has_active_lease(row: OpsRun, *, now: datetime) -> bool:
+    """Return whether ``row`` has a lease strictly after ``now``."""
+    lease_until = row.lease_until
+    if not isinstance(lease_until, datetime):
+        return False
+    if lease_until.tzinfo is None:
+        lease_until = lease_until.replace(tzinfo=timezone.utc)
+    return lease_until > now
+
+
 def max_attempts() -> int:
     try:
         return max(1, int(os.environ.get("OPS_RUN_MAX_ATTEMPTS", "3")))
@@ -163,14 +177,14 @@ async def create_ops_run_from_spec(
 
 async def reopen_stale_claims(session: AsyncSession) -> int:
     """Return claimed rows with expired leases to open."""
-    now = datetime.now(timezone.utc)
+    now = _utcnow()
     stmt = (
         update(OpsRun)
         .where(
             and_(
                 OpsRun.state == "claimed",
                 OpsRun.lease_until.is_not(None),
-                OpsRun.lease_until < now,
+                OpsRun.lease_until <= now,
             )
         )
         .values(
@@ -252,13 +266,15 @@ async def heartbeat_ops_run(
     worker_id: str,
     lease_seconds: int | None = None,
 ) -> OpsRun | None:
-    row = await session.get(OpsRun, run_id)
+    row = await session.get(OpsRun, run_id, with_for_update=True)
     if row is None:
         return None
     if row.state != "claimed" or row.claim_owner != worker_id:
         return None
+    now = _utcnow()
+    if not _has_active_lease(row, now=now):
+        return None
     lease_seconds = lease_seconds or default_lease_seconds()
-    now = datetime.now(timezone.utc)
     row.lease_until = now + timedelta(seconds=lease_seconds)
     row.updated_at = now
     return row
@@ -271,12 +287,14 @@ async def complete_ops_run(
     worker_id: str,
     result: dict[str, Any] | None = None,
 ) -> OpsRun | None:
-    row = await session.get(OpsRun, run_id)
+    row = await session.get(OpsRun, run_id, with_for_update=True)
     if row is None:
         return None
     if row.state != "claimed" or row.claim_owner != worker_id:
         return None
-    now = datetime.now(timezone.utc)
+    now = _utcnow()
+    if not _has_active_lease(row, now=now):
+        return None
     row.state = "succeeded"
     row.lease_until = None
     row.result = normalise_ops_result(result)
@@ -293,12 +311,14 @@ async def fail_ops_run(
     reopen: bool = False,
     result: dict[str, Any] | None = None,
 ) -> OpsRun | None:
-    row = await session.get(OpsRun, run_id)
+    row = await session.get(OpsRun, run_id, with_for_update=True)
     if row is None:
         return None
     if row.state != "claimed" or row.claim_owner != worker_id:
         return None
-    now = datetime.now(timezone.utc)
+    now = _utcnow()
+    if not _has_active_lease(row, now=now):
+        return None
     payload = normalise_ops_result(result)
     if error:
         payload["error"] = error[:2000]

src/activity_core/ops_runs_api.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import hmac
 import os
 import uuid
 from datetime import datetime
@@ -16,6 +17,7 @@ from activity_core.ops_auth import (
     extract_operator_token,
     extract_sso_principal,
     operator_token_configured,
+    require_operator,
 )
 from activity_core.ops_run_queue import (
     claim_ops_runs,
@@ -34,6 +36,7 @@ router = APIRouter(prefix="/ops-runs", tags=["ops-runs"])
 _get_db: Callable[[], async_sessionmaker[AsyncSession]] | None = None
 
 WORKER_TOKEN_ENV = "ACTIVITY_CORE_WORKER_TOKEN"
+WORKER_ID_ENV = "ACTIVITY_CORE_WORKER_ID"
 
 
 def bind_ops_runs_deps(
@@ -53,6 +56,11 @@ def _worker_token_configured() -> bool:
     return bool((os.environ.get(WORKER_TOKEN_ENV) or "").strip())
 
 
+def configured_worker_id() -> str | None:
+    value = (os.environ.get(WORKER_ID_ENV) or "").strip()
+    return value or None
+
+
 def _extract_worker_token(
     *,
     x_worker_token: str | None,
@@ -77,8 +85,18 @@ def require_worker_or_operator(
         x_worker_token=x_worker_token, authorization=authorization
     )
     expected_worker = (os.environ.get(WORKER_TOKEN_ENV) or "").strip()
-    if expected_worker and worker_tok and worker_tok == expected_worker:
+    if (
-        return f"worker:{worker_tok[:8]}"
+        expected_worker
+        and worker_tok
+        and hmac.compare_digest(worker_tok, expected_worker)
+    ):
+        worker_id = configured_worker_id()
+        if not worker_id:
+            raise HTTPException(
+                status_code=503,
+                detail=f"worker identity not configured ({WORKER_ID_ENV})",
+            )
+        return f"worker:{worker_id}"
 
     sso = extract_sso_principal(request)
     if sso:
@@ -88,18 +106,61 @@ def require_worker_or_operator(
         x_operator_token=x_operator_token, authorization=authorization
     )
     expected_op = (os.environ.get("ACTIVITY_CORE_OPERATOR_TOKEN") or "").strip()
-    if expected_op and op_tok and op_tok == expected_op:
+    if expected_op and op_tok and hmac.compare_digest(op_tok, expected_op):
         return "operator:token"
 
     if not _worker_token_configured() and not operator_token_configured():
         if allow_unauth_mutations():
             return "dev:unauth"
-        # Dev convenience when no tokens configured at all
+
-        return "dev:open"
+        raise HTTPException(status_code=503, detail="ops run auth not configured")
 
     raise HTTPException(status_code=401, detail="worker or operator auth required")
 
 
+def require_worker(
+    request: Request,
+    *,
+    x_worker_token: str | None = None,
+    authorization: str | None = None,
+) -> str:
+    """Return the configured worker identity for a valid worker credential."""
+    del request  # reserved for future mTLS/proxy-bound worker principals
+    expected_worker = (os.environ.get(WORKER_TOKEN_ENV) or "").strip()
+    worker_tok = _extract_worker_token(
+        x_worker_token=x_worker_token, authorization=authorization
+    )
+
+    if expected_worker:
+        if not worker_tok or not hmac.compare_digest(worker_tok, expected_worker):
+            raise HTTPException(status_code=401, detail="worker auth required")
+        worker_id = configured_worker_id()
+        if not worker_id:
+            raise HTTPException(
+                status_code=503,
+                detail=f"worker identity not configured ({WORKER_ID_ENV})",
+            )
+        return worker_id
+
+    if allow_unauth_mutations() and not operator_token_configured():
+        return "dev:unauth"
+
+    raise HTTPException(status_code=503, detail="worker auth not configured")
+
+
+def bind_worker_id(requested_worker_id: str, authenticated_worker_id: str) -> str:
+    """Bind a body compatibility field to its authenticated queue principal."""
+    requested = requested_worker_id.strip()
+    if authenticated_worker_id == "dev:unauth":
+        return requested
+    if requested != authenticated_worker_id:
+        raise HTTPException(
+            status_code=403,
+            detail="worker_id does not match authenticated worker",
+        )
+    return authenticated_worker_id
+
+
 class ClaimBody(BaseModel):
     worker_id: str = Field(..., min_length=1, max_length=256)
     labels: list[str] | None = None
@@ -195,21 +256,20 @@ async def post_claim(
     body: ClaimBody,
     request: Request,
     x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
-    x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
     authorization: str | None = Header(default=None),
 ) -> dict[str, Any]:
-    require_worker_or_operator(
+    authenticated_worker_id = require_worker(
         request,
         x_worker_token=x_worker_token,
-        x_operator_token=x_operator_token,
         authorization=authorization,
     )
+    worker_id = bind_worker_id(body.worker_id, authenticated_worker_id)
     Session = _db()
     async with Session() as session:
         async with session.begin():
             claimed = await claim_ops_runs(
                 session,
-                worker_id=body.worker_id,
+                worker_id=worker_id,
                 labels=body.labels,
                 labels_mode=body.labels_mode,
                 limit=body.limit,
@@ -227,28 +287,27 @@ async def post_heartbeat(
     body: HeartbeatBody,
     request: Request,
     x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
-    x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
     authorization: str | None = Header(default=None),
 ) -> dict[str, Any]:
-    require_worker_or_operator(
+    authenticated_worker_id = require_worker(
         request,
         x_worker_token=x_worker_token,
-        x_operator_token=x_operator_token,
         authorization=authorization,
     )
+    worker_id = bind_worker_id(body.worker_id, authenticated_worker_id)
     Session = _db()
     async with Session() as session:
         async with session.begin():
             row = await heartbeat_ops_run(
                 session,
                 run_id,
-                worker_id=body.worker_id,
+                worker_id=worker_id,
                 lease_seconds=body.lease_seconds,
             )
             if row is None:
                 raise HTTPException(
                     status_code=409,
-                    detail="not claimed by this worker or not found",
+                    detail="not actively leased by this worker or not found",
                 )
             return ops_run_to_dict(row)
 
@@ -259,28 +318,27 @@ async def post_complete(
     body: CompleteBody,
     request: Request,
     x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
-    x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
     authorization: str | None = Header(default=None),
 ) -> dict[str, Any]:
-    require_worker_or_operator(
+    authenticated_worker_id = require_worker(
         request,
         x_worker_token=x_worker_token,
-        x_operator_token=x_operator_token,
         authorization=authorization,
     )
+    worker_id = bind_worker_id(body.worker_id, authenticated_worker_id)
     Session = _db()
     async with Session() as session:
         async with session.begin():
             row = await complete_ops_run(
                 session,
                 run_id,
-                worker_id=body.worker_id,
+                worker_id=worker_id,
                 result=body.result,
             )
             if row is None:
                 raise HTTPException(
                     status_code=409,
-                    detail="not claimed by this worker or not found",
+                    detail="not actively leased by this worker or not found",
                 )
             return ops_run_to_dict(row)
 
@@ -291,22 +349,21 @@ async def post_fail(
     body: FailBody,
     request: Request,
     x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
-    x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
     authorization: str | None = Header(default=None),
 ) -> dict[str, Any]:
-    require_worker_or_operator(
+    authenticated_worker_id = require_worker(
         request,
         x_worker_token=x_worker_token,
-        x_operator_token=x_operator_token,
         authorization=authorization,
     )
+    worker_id = bind_worker_id(body.worker_id, authenticated_worker_id)
     Session = _db()
     async with Session() as session:
         async with session.begin():
             row = await fail_ops_run(
                 session,
                 run_id,
-                worker_id=body.worker_id,
+                worker_id=worker_id,
                 error=body.error,
                 reopen=body.reopen,
                 result=body.result,
@@ -314,7 +371,7 @@ async def post_fail(
             if row is None:
                 raise HTTPException(
                     status_code=409,
-                    detail="not claimed by this worker or not found",
+                    detail="not actively leased by this worker or not found",
                 )
             return ops_run_to_dict(row)
 
@@ -322,13 +379,11 @@ async def post_fail(
 @router.post("/expire-leases")
 async def post_expire_leases(
     request: Request,
-    x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
     x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
     authorization: str | None = Header(default=None),
 ) -> dict[str, Any]:
-    require_worker_or_operator(
+    await require_operator(
         request,
-        x_worker_token=x_worker_token,
         x_operator_token=x_operator_token,
         authorization=authorization,
     )

tests/test_ops_run_queue.py

@@ -232,6 +232,7 @@ async def test_complete_persists_only_normalized_glas_evidence() -> None:
     row = MagicMock()
     row.state = "claimed"
     row.claim_owner = "worker-1"
+    row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1)
     row.result = {}
     session = AsyncMock()
     session.get = AsyncMock(return_value=row)
@@ -275,6 +276,7 @@ async def test_fail_reopen_under_max_attempts(monkeypatch: pytest.MonkeyPatch) -
     row.id = uuid.uuid4()
     row.state = "claimed"
     row.claim_owner = "w1"
+    row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1)
     row.attempt = 1
     row.result = {}
     session = AsyncMock()
@@ -297,6 +299,7 @@ async def test_fail_permanent_at_max_attempts(monkeypatch: pytest.MonkeyPatch) -
     row.id = uuid.uuid4()
     row.state = "claimed"
     row.claim_owner = "w1"
+    row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1)
     row.attempt = 3
     row.result = {}
     session = AsyncMock()
@@ -316,6 +319,7 @@ async def test_fail_persists_redacted_failure_evidence() -> None:
     row = MagicMock()
     row.state = "claimed"
     row.claim_owner = "worker-1"
+    row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1)
     row.attempt = 1
     row.result = {}
     session = AsyncMock()
@@ -347,6 +351,87 @@ async def test_fail_persists_redacted_failure_evidence() -> None:
     assert "tool_error" not in row.result
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("lease_offset", [None, timedelta(0), timedelta(seconds=-1)])
+async def test_heartbeat_rejects_missing_or_expired_lease(
+    lease_offset: timedelta | None,
+) -> None:
+    from activity_core import ops_run_queue as oq
+
+    now = datetime(2026, 8, 23, 10, 0, tzinfo=timezone.utc)
+    row = MagicMock()
+    row.state = "claimed"
+    row.claim_owner = "worker-1"
+    row.lease_until = now + lease_offset if lease_offset is not None else None
+    session = AsyncMock()
+    session.get = AsyncMock(return_value=row)
+
+    with patch.object(oq, "_utcnow", return_value=now):
+        heartbeat = await oq.heartbeat_ops_run(
+            session,
+            uuid.uuid4(),
+            worker_id="worker-1",
+            lease_seconds=60,
+        )
+
+    assert heartbeat is None
+    assert row.lease_until == (now + lease_offset if lease_offset is not None else None)
+    session.get.assert_awaited_once()
+    assert session.get.await_args.kwargs == {"with_for_update": True}
+
+
+@pytest.mark.asyncio
+async def test_complete_rejects_expired_lease_without_mutation() -> None:
+    from activity_core import ops_run_queue as oq
+
+    now = datetime(2026, 8, 23, 10, 0, tzinfo=timezone.utc)
+    row = MagicMock()
+    row.state = "claimed"
+    row.claim_owner = "worker-1"
+    row.lease_until = now - timedelta(microseconds=1)
+    row.result = {"before": True}
+    session = AsyncMock()
+    session.get = AsyncMock(return_value=row)
+
+    with patch.object(oq, "_utcnow", return_value=now):
+        completed = await oq.complete_ops_run(
+            session,
+            uuid.uuid4(),
+            worker_id="worker-1",
+            result={"ok": True},
+        )
+
+    assert completed is None
+    assert row.state == "claimed"
+    assert row.result == {"before": True}
+
+
+@pytest.mark.asyncio
+async def test_fail_rejects_wrong_owner_with_active_lease() -> None:
+    from activity_core import ops_run_queue as oq
+
+    now = datetime(2026, 8, 23, 10, 0, tzinfo=timezone.utc)
+    row = MagicMock()
+    row.state = "claimed"
+    row.claim_owner = "worker-1"
+    row.lease_until = now + timedelta(minutes=1)
+    row.result = {"before": True}
+    session = AsyncMock()
+    session.get = AsyncMock(return_value=row)
+
+    with patch.object(oq, "_utcnow", return_value=now):
+        failed = await oq.fail_ops_run(
+            session,
+            uuid.uuid4(),
+            worker_id="worker-2",
+            error="must not persist",
+        )
+
+    assert failed is None
+    assert row.state == "claimed"
+    assert row.result == {"before": True}
+
+
 def test_label_filter_any_vs_all() -> None:
     """Document labels_mode semantics used by claim_ops_runs."""
     row_labels = {"automated", "research-brief"}

tests/test_ops_runs_api.py

@@ -0,0 +1,142 @@
+"""Authentication boundary tests for the ops_run worker API."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import FastAPI, HTTPException
+from fastapi.testclient import TestClient
+
+from activity_core.ops_runs_api import (
+    bind_worker_id,
+    require_worker,
+    require_worker_or_operator,
+    router,
+)
+
+
+def _request(headers: dict[str, str] | None = None) -> MagicMock:
+    request = MagicMock()
+    request.headers = headers or {}
+    return request
+
+
+def _client() -> TestClient:
+    app = FastAPI()
+    app.include_router(router)
+    return TestClient(app)
+
+
+def test_worker_token_binds_configured_identity(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret")
+    monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01")
+
+    authenticated = require_worker(
+        _request(),
+        x_worker_token="worker-secret",
+    )
+
+    assert authenticated == "rein-aharness@railiance01"
+    assert bind_worker_id("rein-aharness@railiance01", authenticated) == authenticated
+
+
+def test_spoofed_worker_id_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret")
+    monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01")
+    authenticated = require_worker(_request(), x_worker_token="worker-secret")
+
+    with pytest.raises(HTTPException) as exc:
+        bind_worker_id("another-worker", authenticated)
+
+    assert exc.value.status_code == 403
+
+
+def test_worker_token_without_identity_fails_closed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret")
+    monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False)
+
+    with pytest.raises(HTTPException) as exc:
+        require_worker(_request(), x_worker_token="worker-secret")
+
+    assert exc.value.status_code == 503
+    assert "ACTIVITY_CORE_WORKER_ID" in exc.value.detail
+
+
+def test_invalid_worker_token_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret")
+    monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01")
+
+    with pytest.raises(HTTPException) as exc:
+        require_worker(_request(), x_worker_token="wrong")
+
+    assert exc.value.status_code == 401
+
+
+def test_operator_or_sso_is_not_worker_identity(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("ACTIVITY_CORE_WORKER_TOKEN", raising=False)
+    monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False)
+    monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "operator-secret")
+
+    for request, kwargs in (
+        (_request({"Remote-User": "alice"}), {}),
+        (_request(), {"authorization": "Bearer operator-secret"}),
+    ):
+        with pytest.raises(HTTPException) as exc:
+            require_worker(request, **kwargs)
+        assert exc.value.status_code == 503
+
+
+def test_claim_endpoint_rejects_operator_and_spoofed_identity(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret")
+    monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01")
+    monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "operator-secret")
+    client = _client()
+    body = {"worker_id": "rein-aharness@railiance01", "limit": 1}
+
+    operator = client.post(
+        "/ops-runs/claim",
+        json=body,
+        headers={"X-Operator-Token": "operator-secret"},
+    )
+    spoofed = client.post(
+        "/ops-runs/claim",
+        json={"worker_id": "another-worker", "limit": 1},
+        headers={"X-Worker-Token": "worker-secret"},
+    )
+
+    assert operator.status_code == 401
+    assert spoofed.status_code == 403
+
+
+def test_unauthenticated_dev_worker_requires_explicit_opt_in(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("ACTIVITY_CORE_WORKER_TOKEN", raising=False)
+    monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False)
+    monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False)
+    monkeypatch.delenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", raising=False)
+
+    with pytest.raises(HTTPException) as exc:
+        require_worker(_request())
+    assert exc.value.status_code == 503
+
+    monkeypatch.setenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", "true")
+    assert require_worker(_request()) == "dev:unauth"
+    assert bind_worker_id("local-worker", "dev:unauth") == "local-worker"
+
+
+def test_read_auth_no_longer_defaults_open(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("ACTIVITY_CORE_WORKER_TOKEN", raising=False)
+    monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False)
+    monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False)
+    monkeypatch.delenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", raising=False)
+
+    with pytest.raises(HTTPException) as exc:
+        require_worker_or_operator(_request())
+
+    assert exc.value.status_code == 503

workplans/ACTIVITY-WP-0036-queue-identity-and-lease-integrity.md

@@ -0,0 +1,135 @@
+---
+id: ACTIVITY-WP-0036
+type: workplan
+title: "Bind queue mutations to worker identity and active leases"
+domain: infotech
+repo: activity-core
+status: active
+owner: codex
+topic_slug: activity-core
+priority: high
+created: "2026-08-23"
+updated: "2026-08-23"
+related:
+  - ACT-ADR-005
+  - ACTIVITY-WP-0026
+  - ACTIVITY-WP-0032
+  - HARNESS-WP-0003
+---
+
+# Bind Queue Mutations to Worker Identity and Active Leases
+
+## Origin
+
+Activity Core's review of rein-aharness ADR-002 found two enforcement gaps at
+the `ops_run` boundary:
+
+1. `ACTIVITY_CORE_WORKER_TOKEN` authenticates a shared caller class, but the
+   authenticated principal is discarded and the caller supplies any
+   `worker_id` in the request body.
+2. heartbeat, completion, and failure accept an owned `claimed` row even when
+   its lease has expired, until another claim or explicit expiry request happens
+   to reopen it.
+
+These gaps make queue ownership advisory at precisely the point where a
+repository executor needs a durable acceptance decision. Rein-side
+cancellation remains necessary, but cannot replace server-side enforcement.
+
+## Boundary and rollout rules
+
+- Activity Core authenticates queue callers and owns lease acceptance.
+- The queue `worker_id` is separate from the governed actor (`agt`) used by
+  Glas and sand-boxer.
+- Operator SSO and break-glass credentials are not worker identities and must
+  not claim, heartbeat, complete, or fail runs through the normal worker API.
+- Existing result normalization and retry ceilings remain unchanged.
+- Production must receive the configured worker identity in the same rollout
+  that activates strict binding; do not strand the live consumer between API
+  and configuration revisions.
+
+## Confirm the cross-repo responsibility contract
+
+```task
+id: ACTIVITY-WP-0036-T01
+status: done
+priority: high
+```
+
+Review rein-aharness ADR-002 against the implemented Activity Core boundary.
+Report any distinction between current state and target guarantees, and require
+the responsibility matrix to assign both worker authentication and lease
+acceptance to an enforceable owner.
+
+Activity Core sent required edits in State Hub message
+`428abb02-75dc-450d-a7f6-56fc715409ab`: the ADR must disclose the current
+self-asserted identity, require credential-to-worker binding, require
+server-side rejection after lease expiry, and keep operator identity out of the
+normal worker mutation path.
+
+## Reject mutations without an active lease
+
+```task
+id: ACTIVITY-WP-0036-T02
+status: done
+priority: high
+```
+
+Lock the target row while deciding heartbeat, completion, or failure. Require
+the row to be claimed by the caller and `lease_until` to be strictly later than
+the server's current UTC time. A missing lease, or a deadline equal to or before
+the decision time, is a conflict and must not update result or state. Stale
+claim reopening must use the same deadline boundary.
+
+Done when unit/API tests cover missing, equal, expired, active, and wrong-owner
+leases and prove no late completion or heartbeat can revive an expired claim.
+
+Implemented with row-level locking and one strict server-time predicate shared
+by heartbeat, completion, and failure. Stale reopening now uses the same
+inclusive expiry boundary (`lease_until <= now`). Tests prove missing, equal,
+expired, active, and wrong-owner behavior without mutating late rows.
+
+## Bind authenticated worker credentials to the claim owner
+
+```task
+id: ACTIVITY-WP-0036-T03
+status: done
+priority: high
+```
+
+Add an explicit configured queue worker identity and bind a valid worker token
+to that identity. Compare the body `worker_id` to the authenticated identity
+before every worker mutation and persist only the authenticated identity as
+`claim_owner`. Refuse production worker mutations when token or identity
+configuration is incomplete. Preserve an explicitly enabled local-development
+path without presenting it as authenticated production behavior.
+
+Done when tests prove spoofed worker ids, operator/SSO credentials on worker
+mutations, missing production identity configuration, and invalid worker
+tokens fail closed.
+
+Implemented `ACTIVITY_CORE_WORKER_ID` as the non-secret identity bound to the
+worker token. Claim and close paths validate the compatibility body field and
+persist only the authenticated identity. Operator/SSO credentials are removed
+from normal worker mutations; unauthenticated development requires the
+existing explicit opt-in, and the former implicit open mode is gone.
+
+## Document, deploy, and prove the boundary
+
+```task
+id: ACTIVITY-WP-0036-T04
+status: progress
+priority: high
+```
+
+Update the queue contract, runbook, and Railiance manifest with the non-secret
+configured worker identity. Run focused and full tests, deploy API and consumer
+configuration without a compatibility gap, and prove one authenticated claim
+plus active heartbeat and a rejected mismatched/expired mutation. Return the
+revision and bounded evidence to rein-aharness for its ADR acknowledgement
+cycle.
+
+Source verification: 468 tests passed with one live integration test skipped;
+Python compilation, whitespace checks, and parsing all 16 Railiance Kubernetes
+documents passed. The manifest binds the existing production worker token to
+`rein-aharness@railiance01`. Production rollout and live rejection evidence
+remain before this task and workplan can finish.