fix: address CodeRabbit review for session control

.agents/skills/agent-release-gate/resources/qa_product.py

@@ -3167,12 +3167,19 @@ def _load_session_control_result(path: str) -> dict:
     skipped = sorted(name for name, status in statuses.items() if status == "SKIP")
     return {
         "path": str(result_path),
-        "status": "FAIL" if failed else "PASS",
+        "status": "FAIL" if failed else ("INCOMPLETE" if skipped else "PASS"),
         "failed": failed,
         "skipped": skipped,
     }
 
 
+def _session_control_result_label(result: dict) -> str:
+    label = f"recorded {result['status']}"
+    if result["skipped"]:
+        label += "; SKIPPED, UNTESTED: " + ", ".join(result["skipped"])
+    return label
+
+
 def main() -> int:
     # Declared here, not beside the assignments below, because the flag help strings read these
     # module defaults and a `global` statement must precede every use of the name in a function.
@@ -3453,7 +3460,7 @@ def main() -> int:
                     "MISSING — no such cell exists"
                     if cell in missing_cells
                     else (
-                        f"recorded {session_control_result['status']}"
+                        _session_control_result_label(session_control_result)
                         if cell == "session_control.py" and session_control_result
                         else "run it separately"
                     )
@@ -3558,7 +3565,7 @@ def main() -> int:
             if cell in CELLS:
                 here = "yes"
             elif cell == "session_control.py" and session_control_result:
-                here = f"recorded {session_control_result['status']}"
+                here = _session_control_result_label(session_control_result)
             else:
                 here = "no — run it separately"
             table += f"| {cell} | {here} | {', '.join(why)} |\n"
@@ -3593,7 +3600,7 @@ def main() -> int:
         for journey in cell["journeys"].values()
     )
     standalone_failed = bool(
-        session_control_result and session_control_result["status"] == "FAIL"
+        session_control_result and session_control_result["status"] != "PASS"
     )
     return 1 if failed or standalone_failed else 0
 

.agents/skills/agent-release-gate/resources/session_control.py

@@ -2098,8 +2098,10 @@ def _judge_stop_approval(evidence: dict, *, pending_found: bool) -> dict:
             "pass --durable-stop on or off"
         )
     late = evidence["late_answer"]
-    if durable_stop == "on" and late.get("status") == 200:
-        return _fail("the late approval answer was accepted after the Stop settled it")
+    if durable_stop == "on" and late.get("status") != 409:
+        return _fail(
+            f"the late approval answer returned HTTP {late.get('status')}, expected 409"
+        )
     if durable_stop == "off":
         if late.get("status") != 200:
             return _fail(

.agents/skills/agent-release-gate/resources/test_qa_product_concurrency.py

@@ -1,12 +1,12 @@
 # /// script
 # requires-python = ">=3.10"
-# dependencies = ["httpx>=0.27"]
+# dependencies = ["httpx>=0.27", "pytest>=8"]
 # ///
 """Offline tests for the `burst` and `crosstalk` journeys. No deployment, no network.
 
 Run either way:
 
-    uv run test_qa_product_concurrency.py     # standalone, prints a line per case
+    uv run test_qa_product_concurrency.py     # standalone, runs through pytest
     uv run --no-sync pytest test_qa_product_concurrency.py
 
 Every case fakes the wire. `invoke` is replaced with a function that builds a `Turn` by hand, so
@@ -436,6 +436,19 @@ def test_session_control_result_consumer_carries_a_failure(tmp_path):
     assert result["failed"] == ["stop-warm"]
 
 
+def test_session_control_result_consumer_marks_skips_incomplete(tmp_path):
+    path = tmp_path / "results.json"
+    path.write_text(json.dumps(_session_control_result("SKIP")))
+
+    result = qa._load_session_control_result(str(path))
+
+    assert result["status"] == "INCOMPLETE"
+    assert result["skipped"]
+    label = qa._session_control_result_label(result)
+    assert "SKIPPED, UNTESTED" in label
+    assert result["skipped"][0] in label
+
+
 def test_session_control_result_consumer_rejects_an_incomplete_run(tmp_path):
     payload = _session_control_result()
     del payload["cells"]["stop-warm"]
@@ -495,6 +508,30 @@ def test_driver_fails_for_a_failed_session_control_result(monkeypatch, tmp_path)
     assert qa.main() == 1
 
 
+def test_driver_fails_for_a_skipped_session_control_result(monkeypatch, tmp_path):
+    result_path = tmp_path / "session-control-results.json"
+    result_path.write_text(json.dumps(_session_control_result("SKIP")))
+    monkeypatch.setattr(qa, "RUNS", tmp_path / "runs")
+    monkeypatch.setitem(qa.JOURNEYS, "chat", lambda _cell: {"pass": True, "why": "ok"})
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "qa_product.py",
+            "--cell",
+            "C3",
+            "--only",
+            "chat",
+            "--changed-path",
+            "api/oss/src/core/sessions/service.py",
+            "--session-control-results",
+            str(result_path),
+        ],
+    )
+
+    assert qa.main() == 1
+
+
 def test_the_driver_forces_a_mandatory_journey_past_only(tmp_path=None):
     """End to end through main(), with every journey stubbed out."""
     import tempfile
@@ -958,12 +995,7 @@ def never_ends(session, messages, params, timeout=300.0, deadline=None):
 
 
 def main() -> int:
-    cases = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
-    for case in cases:
-        case()
-        print(f"PASS {case.__name__}")
-    print(f"\n{len(cases)} offline cases passed")
-    return 0
+    return pytest.main([__file__, "-q"])
 
 
 if __name__ == "__main__":

.agents/skills/agent-release-gate/resources/test_session_control.py

@@ -85,10 +85,13 @@ def _stop_approval_evidence(*, durable_stop: str, late_status: int) -> dict:
 
 
 def test_stop_approval_durable_path_requires_late_answer_refusal():
-    accepted = _stop_approval_evidence(durable_stop="on", late_status=200)
     refused = _stop_approval_evidence(durable_stop="on", late_status=409)
 
-    assert sc._judge_stop_approval(accepted, pending_found=True)["pass"] is False
+    for status in (200, 202, 500):
+        unexpected = _stop_approval_evidence(durable_stop="on", late_status=status)
+        verdict = sc._judge_stop_approval(unexpected, pending_found=True)
+        assert verdict["pass"] is False
+        assert f"HTTP {status}, expected 409" in verdict["why"]
     verdict = sc._judge_stop_approval(refused, pending_found=True)
     assert verdict["pass"] is True
     assert "late answer was refused" in verdict["why"]

api/oss/src/apis/fastapi/sessions/router.py

@@ -551,6 +551,9 @@ async def heartbeat_session_stream(
         if not has_permission:
             raise FORBIDDEN_EXCEPTION
 
+        if payload.release_owner:
+            _assert_runner_token(request)
+
         heartbeat = await self._service.heartbeat(
             project_id=project_id,
             request=payload,

api/oss/src/utils/env.py

@@ -537,11 +537,21 @@ class SessionsRecordsConfig(BaseModel):
 
     # How long a record message the worker failed to write sits unacknowledged before the
     # worker claims it back and tries again.
-    reclaim_idle_ms: int = int(os.getenv("AGENTA_RECORDS_RECLAIM_IDLE_MS") or 30_000)
+    reclaim_idle_ms: int = Field(
+        default_factory=lambda: int(
+            os.getenv("AGENTA_RECORDS_RECLAIM_IDLE_MS") or 30_000
+        ),
+        ge=0,
+        validate_default=True,
+    )
 
     # Deliveries after which a record message is dropped instead of retried forever. A message
     # Postgres never accepts would otherwise hold every later message in the group.
-    max_deliveries: int = int(os.getenv("AGENTA_RECORDS_MAX_DELIVERIES") or 5)
+    max_deliveries: int = Field(
+        default_factory=lambda: int(os.getenv("AGENTA_RECORDS_MAX_DELIVERIES") or 5),
+        ge=1,
+        validate_default=True,
+    )
 
     model_config = ConfigDict(extra="ignore")
 

api/oss/tests/pytest/unit/sessions/test_heartbeat_release_auth.py

@@ -0,0 +1,97 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import UUID
+
+import pytest
+from fastapi import HTTPException
+
+from oss.src.apis.fastapi.sessions import router as router_module
+from oss.src.apis.fastapi.sessions.router import SessionStreamsRouter
+from oss.src.core.sessions.streams.dtos import (
+    SessionHeartbeatRequest,
+    SessionHeartbeatResult,
+)
+from oss.src.utils.env import env
+
+
+_PROJECT = UUID("00000000-0000-0000-0000-0000000000aa")
+_USER = UUID("00000000-0000-0000-0000-0000000000bb")
+
+
+def _request(headers=None):
+    return SimpleNamespace(
+        state=SimpleNamespace(project_id=_PROJECT, user_id=_USER),
+        headers=headers or {},
+    )
+
+
+def _router(service):
+    return SessionStreamsRouter(
+        service=service,
+        interactions_service=SimpleNamespace(),
+    )
+
+
+@pytest.mark.asyncio
+async def test_release_owner_heartbeat_requires_the_runner_token(monkeypatch):
+    monkeypatch.setattr(env.runner, "token", "runner-secret")
+    monkeypatch.setattr(
+        router_module, "check_action_access", AsyncMock(return_value=True)
+    )
+    service = SimpleNamespace(heartbeat=AsyncMock())
+
+    with pytest.raises(HTTPException) as exc_info:
+        await _router(service).heartbeat_session_stream(
+            _request(),
+            SessionHeartbeatRequest(
+                session_id="session-1",
+                replica_id="replica-1",
+                release_owner=True,
+            ),
+        )
+
+    assert exc_info.value.status_code == 401
+    service.heartbeat.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_regular_heartbeat_keeps_user_authentication_only(monkeypatch):
+    monkeypatch.setattr(env.runner, "token", "runner-secret")
+    monkeypatch.setattr(
+        router_module, "check_action_access", AsyncMock(return_value=True)
+    )
+    service = SimpleNamespace(
+        heartbeat=AsyncMock(return_value=SessionHeartbeatResult(replica_id="replica-1"))
+    )
+    payload = SessionHeartbeatRequest(
+        session_id="session-1",
+        replica_id="replica-1",
+        turn_id="turn-1",
+    )
+
+    result = await _router(service).heartbeat_session_stream(_request(), payload)
+
+    assert result.replica_id == "replica-1"
+    service.heartbeat.assert_awaited_once_with(project_id=_PROJECT, request=payload)
+
+
+@pytest.mark.asyncio
+async def test_release_owner_accepts_the_shared_runner_token(monkeypatch):
+    monkeypatch.setattr(env.runner, "token", "runner-secret")
+    monkeypatch.setattr(
+        router_module, "check_action_access", AsyncMock(return_value=True)
+    )
+    service = SimpleNamespace(
+        heartbeat=AsyncMock(return_value=SessionHeartbeatResult(replica_id="replica-1"))
+    )
+    payload = SessionHeartbeatRequest(
+        session_id="session-1",
+        replica_id="replica-1",
+        release_owner=True,
+    )
+
+    await _router(service).heartbeat_session_stream(
+        _request({"X-Agenta-Runner-Token": "runner-secret"}), payload
+    )
+
+    service.heartbeat.assert_awaited_once_with(project_id=_PROJECT, request=payload)

api/oss/tests/pytest/unit/sessions/test_records_config.py

@@ -0,0 +1,36 @@
+import pytest
+from pydantic import ValidationError
+
+from oss.src.utils.env import SessionsRecordsConfig
+
+
+def test_session_record_retry_bounds_accept_the_minimum_values():
+    config = SessionsRecordsConfig(reclaim_idle_ms=0, max_deliveries=1)
+
+    assert config.reclaim_idle_ms == 0
+    assert config.max_deliveries == 1
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [("reclaim_idle_ms", -1), ("max_deliveries", 0), ("max_deliveries", -1)],
+)
+def test_session_record_retry_bounds_reject_invalid_values(field, value):
+    with pytest.raises(ValidationError):
+        SessionsRecordsConfig(**{field: value})
+
+
+@pytest.mark.parametrize(
+    ("name", "value"),
+    [
+        ("AGENTA_RECORDS_RECLAIM_IDLE_MS", "-1"),
+        ("AGENTA_RECORDS_MAX_DELIVERIES", "0"),
+    ],
+)
+def test_session_record_retry_bounds_validate_environment_defaults(
+    monkeypatch, name, value
+):
+    monkeypatch.setenv(name, value)
+
+    with pytest.raises(ValidationError):
+        SessionsRecordsConfig()

docs/design/session-control-and-live-events/research.md

@@ -21,10 +21,15 @@ when a heartbeat returns `is_current_turn=false`, then aborts locally.
 `DELETE /sessions/streams/?session_id=...` is separate from normal Cancel. It contacts the runner
 and tears down the sandbox. The session remains resumable after Cancel but not after Kill.
 
-The direct kill client uses one configured `runner.internal_url`. Redis separately stores the
-logical owner `replica_id`. The current kill client does not resolve that identifier to a
-replica-specific address. Immediate Cancel cannot assume that logical owner identity already
-provides direct network routing.
+The v1 direct client uses one configured `runner.internal_url`. It is correct for a single runner,
+or when the URL fronts an owner-aware router. Redis separately stores the logical owner
+`replica_id`, but the direct client does not resolve that identity to a replica-specific address.
+A request that reaches the wrong replica returns not found and must not be treated as success.
+
+Immediate Cancel remains durable, so an unavailable owner can recover and apply it later or be
+settled as lost. Kill is best effort through the same configured URL. Until owner-aware forwarding
+exists, a multi-runner Kill cannot guarantee immediate teardown; authoritative session state is
+cleared and sandbox lease or orphan cleanup provides the fallback.
 
 ### Heartbeat
 

docs/design/session-control-and-live-events/rfc.md

@@ -44,7 +44,7 @@ Idempotency-Key: <client-generated-key>
 {
   "type": "send",
   "message": "Explain this failure",
-  "delivery": "reject"
+  "on_busy": "reject"
 }
 ```
 
@@ -126,7 +126,7 @@ execution is already running:
 - `queue`: save the new message. Start it after current work stops normally.
 - `steer`: save the new message. Interrupt current work, then start the new message.
 
-When the session is idle, all accepted messages start normally. The contract may call this field
+When the session is idle, all accepted messages start normally. The contract calls this field
 `on_busy` so its purpose is clear.
 
 ### Visible pending messages
@@ -243,6 +243,10 @@ acknowledges it, and immediately opens the next request. A disconnected runner r
 claims commands that remain durable. Redis or Postgres notifications may wake API replicas
 internally, but the runner never connects to either system.
 
+Credential-bearing long polls require HTTPS with normal certificate validation. The client must
+disable redirects or reject any redirect whose origin differs from the configured API origin, and
+it must never forward runner credentials across origins.
+
 A persistent WebSocket or bidirectional stream can later reduce repeated requests and carry richer
 runner status. It is not required for the first contract. Direct API calls into runner pods and
 per-runner Redis subscriptions are poor fits for user-operated runners because they require inbound
@@ -303,8 +307,10 @@ execution: running -> stopping -> stopped
 
 The API accepts Stop by durably creating the command and moving the matching execution to
 `stopping` in one transaction. `expected_execution_id` remains optional. A command claim has a
-lease and can be delivered again after disconnection. The runner deduplicates by `command_id` and
-validates the execution ID and ownership generation before applying it.
+lease and can be delivered again after disconnection. In a fenced design, the runner deduplicates
+by `command_id` and validates both the execution ID and ownership generation before applying it.
+The v1 direct-delivery adapter has no generation token; it validates the target execution ID and
+requires the addressed runner replica to own that execution.
 
 Claiming or acknowledging a command does not prove that execution stopped. Public clients follow
 execution state. The runner normally reports the terminal outcome and the API settles the command