Bind multiple queue worker identities, one token each (WP-0039-T01)

ACTIVITY_CORE_WORKERS maps worker_id=ENV_NAME, where each token env must be
ACTIVITY_CORE_WORKER_TOKEN[_SUFFIX]. Without the map, the legacy single pair
behaves exactly as before. Duplicate identities, missing or shared tokens, a
token equal to the operator token, and an unlisted legacy identity all fail
worker mutations closed with 503. Operator/SSO reads keep working.

Declare per-identity OpenBao paths and an ExternalSecret, not yet applied.
The policy, seeding and rollout are waiting tasks T02-T04, answering
secrets-engine SECRETS-WP-0009-T03 and SECRETS-WP-0011-T04.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 151606@bnt-lap001
Assistant-Session: 3c0a4ad5-bb8b-4bf7-b9f0-fa5f29204e48

tests/test_ops_runs_api.py

@@ -165,3 +165,113 @@ def test_close_response_distinguishes_reconciled_and_refusal_codes(
         _close_response(CloseOpsRunOutcome("expired_lease", row))
     assert exc.value.status_code == 409
     assert exc.value.detail["code"] == "expired_lease"
+
+
+def _clear_worker_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    for name in (
+        "ACTIVITY_CORE_WORKERS",
+        "ACTIVITY_CORE_WORKER_ID",
+        "ACTIVITY_CORE_WORKER_TOKEN",
+        "ACTIVITY_CORE_WORKER_TOKEN_METERED",
+        "ACTIVITY_CORE_OPERATOR_TOKEN",
+        "ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS",
+    ):
+        monkeypatch.delenv(name, raising=False)
+
+
+def _two_workers(monkeypatch: pytest.MonkeyPatch) -> None:
+    _clear_worker_env(monkeypatch)
+    monkeypatch.setenv(
+        "ACTIVITY_CORE_WORKERS",
+        "rein-aharness@railiance01=ACTIVITY_CORE_WORKER_TOKEN,"
+        "rein-aharness-metered@railiance01=ACTIVITY_CORE_WORKER_TOKEN_METERED",
+    )
+    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "loop-secret")
+    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN_METERED", "metered-secret")
+
+
+def test_each_worker_token_binds_exactly_its_own_identity(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _two_workers(monkeypatch)
+
+    loop = require_worker(_request(), x_worker_token="loop-secret")
+    metered = require_worker(_request(), authorization="Bearer metered-secret")
+
+    assert loop == "rein-aharness@railiance01"
+    assert metered == "rein-aharness-metered@railiance01"
+    with pytest.raises(HTTPException) as exc:
+        bind_worker_id("rein-aharness@railiance01", metered)
+    assert exc.value.status_code == 403
+    with pytest.raises(HTTPException) as exc:
+        require_worker(_request(), x_worker_token="unknown")
+    assert exc.value.status_code == 401
+
+
+def test_claim_endpoint_rejects_cross_identity_body(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _two_workers(monkeypatch)
+
+    response = _client().post(
+        "/ops-runs/claim",
+        json={"worker_id": "rein-aharness@railiance01", "limit": 1},
+        headers={"X-Worker-Token": "metered-secret"},
+    )
+
+    assert response.status_code == 403
+
+
+@pytest.mark.parametrize(
+    ("workers", "env", "fragment"),
+    [
+        ("a=ACTIVITY_CORE_WORKER_TOKEN,b=ACTIVITY_CORE_WORKER_TOKEN_METERED",
+         {"ACTIVITY_CORE_WORKER_TOKEN": "same", "ACTIVITY_CORE_WORKER_TOKEN_METERED": "same"},
+         "shared"),
+        ("a=ACTIVITY_CORE_WORKER_TOKEN,a=ACTIVITY_CORE_WORKER_TOKEN_METERED",
+         {"ACTIVITY_CORE_WORKER_TOKEN": "x", "ACTIVITY_CORE_WORKER_TOKEN_METERED": "y"},
+         "duplicate"),
+        ("a=ACTIVITY_CORE_WORKER_TOKEN,b=ACTIVITY_CORE_WORKER_TOKEN_METERED",
+         {"ACTIVITY_CORE_WORKER_TOKEN": "x"},
+         "ACTIVITY_CORE_WORKER_TOKEN_METERED"),
+        ("a=ACTIVITY_CORE_OPERATOR_TOKEN",
+         {"ACTIVITY_CORE_OPERATOR_TOKEN": "op"},
+         "invalid"),
+        ("a=ACTIVITY_CORE_WORKER_TOKEN",
+         {"ACTIVITY_CORE_WORKER_TOKEN": "op", "ACTIVITY_CORE_OPERATOR_TOKEN": "op"},
+         "operator"),
+        ("a=ACTIVITY_CORE_WORKER_TOKEN",
+         {"ACTIVITY_CORE_WORKER_TOKEN": "x", "ACTIVITY_CORE_WORKER_ID": "b"},
+         "not listed"),
+    ],
+)
+def test_ambiguous_worker_config_fails_closed(
+    monkeypatch: pytest.MonkeyPatch,
+    workers: str,
+    env: dict[str, str],
+    fragment: str,
+) -> None:
+    _clear_worker_env(monkeypatch)
+    monkeypatch.setenv("ACTIVITY_CORE_WORKERS", workers)
+    for name, value in env.items():
+        monkeypatch.setenv(name, value)
+
+    with pytest.raises(HTTPException) as exc:
+        require_worker(_request(), x_worker_token="x")
+
+    assert exc.value.status_code == 503
+    assert fragment in exc.value.detail
+
+
+def test_broken_worker_config_does_not_lock_out_operator(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _clear_worker_env(monkeypatch)
+    monkeypatch.setenv("ACTIVITY_CORE_WORKERS", "a=ACTIVITY_CORE_WORKER_TOKEN_MISSING")
+    monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "operator-secret")
+
+    principal = require_worker_or_operator(
+        _request(), authorization="Bearer operator-secret"
+    )
+
+    assert principal == "operator:token"