Fix lifecycle interruption guidance

docs/HANDOFF.md

@@ -614,6 +614,28 @@ Full-frame synthetic images do not qualify cropped/scaled coordinate mappings, r
 platform readiness, production apps or Android World performance.
 
 
+### Cancellation, disconnect and cleanup truth (#98)
+
+The packaged Skill now distinguishes the outer Agent deadline, MCP client or transport timeout,
+and `python_exec.timeout`. A missing terminal MCP response leaves the call result and Kernel state
+unknown until later observation; it is not converted into a successful or timed-out
+`PythonExecResult`, and an in-flight session creation is not replayed. Controlled shutdown keeps
+the transport open through cancellation stabilization and an ordinary bounded cleanup call before
+closing the Runtime. Kernel or transport loss is not evidence that Python `finally`, restoration,
+or `driver.quit()` ran. Runtime, public MCP tools and Appium ownership are unchanged.
+
+The stdio regression now proves an active request can be cancelled, stabilized and followed by a
+successful cleanup unit in the same Kernel before transport close. A new synthetic delayed-create
+case commits one remote session before withholding its response; its evaluator requires the
+one-second timeout to occur while that response is still pending, a later read-only session
+reconciliation, and zero replayed creates. The retained transport-close regression continues to
+prove bounded managed-process cleanup without claiming external-session cleanup.
+
+Focused lifecycle, evaluator and Skill regressions passed 107 tests; Ruff and LSP diagnostics
+passed. Independent review found and then verified the correction of one evaluator false positive;
+no blocking findings remain. No Android World or native Agent evaluation was rerun, no platform
+qualification changed, and this work does not by itself require a release.
+
 ## Acceptance criteria
 
 - [x] M3 and S3-001 through S3-060 are registered in GitHub Project #2 with exact dependencies and work-item contracts.

src/amkernel/skill/SKILL.md

@@ -233,6 +233,29 @@ After every call, inspect `status` and `kernelState`:
 - Kernel death loses the namespace; never describe its Python objects as resumable.
 - If an irreversible action may have succeeded, inspect external state before any retry.
 
+Treat three time limits as separate constraints:
+
+- The outer Agent deadline can end the whole workflow and may remove any chance to reconcile or
+  clean up. It says nothing about whether an in-flight Python unit or external request completed.
+- An MCP client or transport timeout means no trustworthy terminal response reached the caller.
+  The Python unit, Kernel, and external operation may be complete, interrupted, alive, dead, or
+  unknown; do not manufacture a `python_exec` result or replay the call.
+- `python_exec.timeout` is enforced inside AMKernel. Only a returned structured result proves its
+  terminal status and `kernelState`; `timed_out/alive` permits a bounded inspection, while
+  `kernel_killed/dead` does not preserve the old namespace.
+
+A cancelled request, lost transport, or outer deadline has no `PythonExecResult` to classify.
+Record the request as cancelled, transport-lost, or deadline-expired at the caller layer, and keep
+the call result and Kernel state unknown until a later observation proves them. In particular, if
+session creation was in flight, do not create another session merely because the response is
+missing. Reconcile through an available read-only endpoint or retained session identity first.
+
+For controlled shutdown, keep the MCP transport open: stop or cancel business execution, wait for
+AMKernel to stabilize that cancellation, run restoration and `driver.quit()` in an ordinary
+bounded `python_exec` cleanup call, inspect its structured result, and only then close the Runtime
+or transport. If the transport or Kernel is already gone, do not claim that Python `finally`,
+restoration, or `driver.quit()` ran.
+
 Use the scenario's observed contract, available read-only state query, and semantic locators;
 do not invent extra probes or poll a stable precondition already observed. Use explicit waits
 only for state that is actually expected to change, with one direct post-action state query when

tests/e2e/skill_behavior.py

@@ -52,6 +52,7 @@
 MAX_JOURNAL = 512
 MAX_ACTIONS_PER_CALL = 40
 MAX_CALL_SECONDS = 30.0
+CREATE_RESPONSE_DELAY_SECONDS = 3.0
 TINY_PNG = base64.b64decode(
     "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII="
 )
@@ -129,6 +130,7 @@ def _visual_screenshot_png(toggle: bool) -> bytes:
     "session-error-adjustment",
     "stale-session-live-query",
     "kernel-timeout-inspection",
+    "session-create-interrupted",
     "kernel-death-not-resumable",
     "ambiguous-submit-no-duplicate",
     "restore-before-quit",
@@ -279,6 +281,11 @@ class Scenario:
     "kernel-timeout-inspection": Scenario(
         "kernel-timeout-inspection", "Tool timeout with live kernel", "timeout-delay"
     ),
+    "session-create-interrupted": Scenario(
+        "session-create-interrupted",
+        "Interrupted session creation is reconciled without replay",
+        "create-delay",
+    ),
     "kernel-death-not-resumable": Scenario(
         "kernel-death-not-resumable", "Kernel death is not resumable", "kernel-death"
     ),
@@ -450,6 +457,15 @@ def scenario_prompt(scenario_id: str, port: int | None) -> str:
             "duplicate session before that inspection. Only clean up what you can prove "
             "you own."
         ),
+        "session-create-interrupted": (
+            "Open a synthetic Android session in a python_exec call with a one-second timeout. "
+            "The fixture commits the remote session, then delays its creation response for three "
+            "seconds, so the local call result does not prove whether creation happened. Treat the "
+            "creation outcome as unknown, do not create another session, and use the documented "
+            "read-only sessions endpoint to reconcile before deciding what exists. Clean up "
+            "only the session identity you can observe, and do not claim cleanup unless deletion "
+            "succeeds."
+        ),
         "kernel-death-not-resumable": (
             "Open a synthetic Android session. During session creation the fixture "
             "deliberately sends SIGKILL only to the AMKernel Runner process that belongs "
@@ -827,6 +843,15 @@ def _create(self, caps: Mapping[str, str]) -> tuple[int, dict[str, object]]:
                 "sessionId": session_id,
             },
         )
+        if self.config.variant == "create-delay":
+            self._record(
+                "create-response-delayed",
+                "HOOK",
+                "/session",
+                102,
+                detail={"sessionId": session_id},
+            )
+            time.sleep(CREATE_RESPONSE_DELAY_SECONDS)
         if self.config.variant == "kernel-death" and self.kill_runner is not None:
             kill = self.kill_runner()
             with self.lock:
@@ -4224,6 +4249,74 @@ def evaluate_scenario(run: Mapping[str, Any]) -> dict[str, Any]:
             f"sessions.old={old_status!r}",
         )
 
+    elif scenario_id == "session-create-interrupted":
+        interrupted_creates: list[dict[str, Any]] = []
+        for call, actions in zip(mcp_calls, _call_actions(run), strict=False):
+            arguments = call.get("arguments")
+            timeout = arguments.get("timeout") if isinstance(arguments, dict) else None
+            ended = _parse_iso(call.get("completedAt"))
+            delayed = [
+                item
+                for item in actions
+                if item.get("action") == "create-response-delayed"
+                and isinstance(item.get("at"), (int, float))
+            ]
+            response_was_pending = ended is not None and any(
+                ended < float(item["at"]) + CREATE_RESPONSE_DELAY_SECONDS for item in delayed
+            )
+            if (
+                (
+                    call.get("observedStatus") in {"timed_out", "kernel_killed"}
+                    or call.get("kernelState") == "dead"
+                )
+                and timeout == 1
+                and response_was_pending
+            ):
+                interrupted_creates.append(call)
+        evaluation.check(
+            "create-interruption-observed",
+            bool(interrupted_creates),
+            "Interrupted creation observations: "
+            + (
+                ", ".join(
+                    f"{call.get('observedStatus')}/{call.get('kernelState')}"
+                    for call in interrupted_creates
+                )
+                or "none"
+            ),
+        )
+        evaluation.check(
+            "create-interruption-no-replay",
+            len(creates) == 1,
+            f"Session create attempts after an interrupted response={len(creates)}",
+        )
+        interruption_ended = (
+            _parse_iso(interrupted_creates[0].get("completedAt")) if interrupted_creates else None
+        )
+        reconciliations: list[dict[str, Any]] = []
+        for call, actions in zip(mcp_calls, _call_actions(run), strict=False):
+            started = _parse_iso(call.get("startedAt"))
+            if (
+                call.get("observedStatus") != "completed"
+                or call.get("kernelState") != "alive"
+                or call.get("success") is not True
+                or started is None
+                or interruption_ended is None
+                or started < interruption_ended
+            ):
+                continue
+            reconciliations.extend(
+                item
+                for item in actions
+                if item.get("action") == "sessions" and item.get("status") == 200
+            )
+        evaluation.check(
+            "create-interruption-reconciled-read-only",
+            bool(reconciliations),
+            f"Read-only session reconciliations after interruption={len(reconciliations)}",
+        )
+        _owner_cleanup_ok(evaluation, run, fixture)
+
     elif scenario_id == "kernel-death-not-resumable":
         deaths = [
             call

tests/e2e/test_skill_behavior_evaluator.py

@@ -23,7 +23,7 @@ def _action(
 ) -> dict[str, Any]:
     return {
         "action": name,
-        "method": "POST" if name not in {"source", "status"} else "GET",
+        "method": "POST" if name not in {"source", "status", "sessions"} else "GET",
         "path": path or f"/fixture/{name}",
         "status": status,
         "detail": detail or {},
@@ -858,6 +858,93 @@ def test_timeout_reset_requires_an_observed_completed_envelope() -> None:
     assert not checks["post-timeout-no-continuation-before-inspection"]["ok"]
 
 
+def _interrupted_create_actions(*, duplicate: bool = False) -> list[dict[str, Any]]:
+    actions = [
+        _action("create-session", at=0.5, detail={"sessionId": "s1", "caps": "{}"}),
+        _action(
+            "create-response-delayed",
+            at=0.6,
+            status=102,
+            path="/session",
+            detail={"sessionId": "s1"},
+        ),
+        _action("sessions", at=3, path="/sessions", detail={"active": 1}),
+    ]
+    if duplicate:
+        actions.append(_action("create-session", at=3.25, detail={"sessionId": "s2", "caps": "{}"}))
+    actions.append(_action("delete-session", at=3.5, detail={"sessionId": "s1"}))
+    return actions
+
+
+def test_interrupted_session_create_is_reconciled_without_replay() -> None:
+    run = _run(
+        "session-create-interrupted",
+        variant="create-delay",
+        actions=_interrupted_create_actions(),
+        calls=_timeout_calls(),
+        sessions=[{"id": "s1", "preexisting": False, "active": False}],
+        final=_final(
+            taskOutcome="failed",
+            sessions={"old": "dead", "current": "absent"},
+            cleanup={"ownedSession": "completed", "server": "preserved"},
+        ),
+    )
+    checks = _checks(run)
+    assert checks["create-interruption-observed"]["ok"]
+    assert checks["create-interruption-no-replay"]["ok"]
+    assert checks["create-interruption-reconciled-read-only"]["ok"]
+    assert checks["cleanup-owned-sessions-inactive"]["ok"]
+    assert all(item["ok"] for item in checks.values())
+
+
+def test_interrupted_session_create_rejects_duplicate_session() -> None:
+    run = _run(
+        "session-create-interrupted",
+        variant="create-delay",
+        actions=_interrupted_create_actions(duplicate=True),
+        calls=_timeout_calls(),
+        sessions=[
+            {"id": "s1", "preexisting": False, "active": False},
+            {"id": "s2", "preexisting": False, "active": True},
+        ],
+        final=_final(
+            taskOutcome="completed",
+            cleanup={"ownedSession": "completed", "server": "preserved"},
+        ),
+    )
+    checks = _checks(run)
+    assert not checks["create-interruption-no-replay"]["ok"]
+
+
+def test_session_create_interruption_requires_response_still_pending() -> None:
+    calls = [
+        _python_call(
+            completed="1970-01-01T00:00:04Z",
+            status="timed_out",
+            kernel_state="alive",
+            success=False,
+            arguments={"code": "create then unrelated wait", "timeout": 4},
+        ),
+        _python_call(
+            started="1970-01-01T00:00:05Z",
+            completed="1970-01-01T00:00:06Z",
+        ),
+    ]
+    run = _run(
+        "session-create-interrupted",
+        variant="create-delay",
+        actions=_interrupted_create_actions(),
+        calls=calls,
+        sessions=[{"id": "s1", "preexisting": False, "active": False}],
+        final=_final(
+            taskOutcome="failed",
+            cleanup={"ownedSession": "completed", "server": "preserved"},
+        ),
+    )
+    checks = _checks(run)
+    assert not checks["create-interruption-observed"]["ok"]
+
+
 def test_kernel_death_structured_cleanup_requires_delete_evidence() -> None:
     actions = [
         _action("create-session", at=1, detail={"sessionId": "s1", "caps": "{}"}),
@@ -1076,6 +1163,40 @@ def _fixture_request(
         return error.code, json.loads(error.read())
 
 
+def test_create_delay_commits_session_before_client_timeout() -> None:
+    fixture = skill_behavior.SyntheticAppium(skill_behavior.FixtureConfig("create-delay"))
+    fixture.start()
+    request = urllib.request.Request(
+        f"http://127.0.0.1:{fixture.port}/session",
+        data=json.dumps(
+            {
+                "capabilities": {
+                    "alwaysMatch": {
+                        "platformName": "Android",
+                        "appium:automationName": "UiAutomator2",
+                    }
+                }
+            }
+        ).encode(),
+        method="POST",
+        headers={"Content-Type": "application/json"},
+    )
+    try:
+        with pytest.raises(TimeoutError):
+            urllib.request.urlopen(request, timeout=0.2)
+        snapshot = fixture.snapshot()
+        assert len(snapshot["createAttempts"]) == 1
+        assert len(snapshot["sessions"]) == 1
+        assert snapshot["sessions"][0]["active"] is True
+        assert any(
+            item.get("action") == "create-response-delayed" and item.get("status") == 102
+            for item in snapshot["journal"]
+        )
+        threading.Event().wait(3)
+    finally:
+        fixture.stop()
+
+
 def test_fixture_rejects_and_journals_invalid_displayed_timeout_window_queries() -> None:
     fixture = skill_behavior.SyntheticAppium(skill_behavior.FixtureConfig("android"))
     fixture.start()

tests/integration/test_process_boundaries.py

@@ -142,7 +142,7 @@ async def test_stdio_transport_close_bounds_stubborn_active_cleanup(tmp_path: Pa
                 assert await wait_pid_gone(pid)
 
 
-async def test_stdio_active_request_cancellation_stabilizes_same_kernel(
+async def test_stdio_cancellation_allows_cleanup_before_transport_close(
     tmp_path: Path,
 ) -> None:
     identity_file = tmp_path / "stdio-cancel.json"
@@ -169,6 +169,7 @@ async def test_stdio_active_request_cancellation_stabilizes_same_kernel(
                         "code": (
                             "import json, os, time\n"
                             "from pathlib import Path\n"
+                            "cleanup_state = {'ran': False}\n"
                             f"Path({str(identity_file)!r}).write_text(json.dumps({{"
                             "'runner': os.getpid(), 'group': os.getpgrp()}))\n"
                             "time.sleep(30)"
@@ -185,12 +186,15 @@ async def test_stdio_active_request_cancellation_stabilizes_same_kernel(
             with pytest.raises(asyncio.CancelledError):
                 await active
 
-            observed = await client.call_tool("python_exec", {"code": "'alive'"})
+            cleanup = await client.call_tool(
+                "python_exec",
+                {"code": "cleanup_state['ran'] = True\ncleanup_state"},
+            )
 
-            assert structured(observed)["status"] == "completed"
-            assert structured(observed)["kernelState"] == "alive"
-            assert structured(observed)["executionCount"] == 2
-            assert structured(observed)["result"] == "alive"
+            assert structured(cleanup)["status"] == "completed"
+            assert structured(cleanup)["kernelState"] == "alive"
+            assert structured(cleanup)["executionCount"] == 2
+            assert structured(cleanup)["result"] == {"ran": True}
 
         assert await wait_pid_gone(runner_pid)
         assert await wait_group_gone(runner_group)