ci: exercise deferred agent dispatch and surface background failures

.github/workflows/python-ci.yml

@@ -50,7 +50,7 @@ jobs:
         if: steps.scope.outputs.runtime_changed == 'true'
         run: |
           python -m pip check
-          python -m pytest -q --durations=10 --junitxml=test-results/python.xml
+          python -m pytest -v --durations=10 --junitxml=test-results/python.xml
 
       - name: Upload test results
         if: always() && steps.scope.outputs.runtime_changed == 'true'
@@ -94,7 +94,7 @@ jobs:
       - name: Verify shared leases, startup recovery, and scheduler leadership
         if: steps.scope.outputs.runtime_changed == 'true'
         run: >-
-          python -m pytest -q --durations=10 --junitxml=test-results/windows-lifecycle.xml
+          python -m pytest -v --durations=10 --junitxml=test-results/windows-lifecycle.xml
           tests/application/test_application_lease.py
           tests/application/test_automation_scheduler_leadership.py
           tests/application/test_session_deletion_service.py
@@ -106,7 +106,7 @@ jobs:
       - name: Verify Windows ACLs and Job Object sandbox
         if: steps.scope.outputs.runtime_changed == 'true'
         run: >-
-          python -m pytest -q --durations=10 --junitxml=test-results/windows-platform.xml
+          python -m pytest -v --durations=10 --junitxml=test-results/windows-platform.xml
           tests/test_private_storage_windows.py
           tests/test_harness_sandbox.py
           tests/test_exec_sandbox_wiring.py

docs/CI.md

@@ -64,6 +64,10 @@ test baseline does not replace installation compatibility checks.
 
 Inspect the earliest failing step. Python and Windows jobs upload JUnit reports
 with test names and durations; browser jobs retain failure traces and screenshots.
+Python logs name each test as it runs. If one test takes longer than 60 seconds,
+pytest prints all Python thread stacks; this is diagnostic output, not an extended
+deadline or a successful result. Unhandled background-thread exceptions and
+unraisable exceptions fail the test instead of appearing only as warnings.
 Packaged startup errors include worker output. A failure while stopping the test
 service is attached to the original exception, and temporary cleanup is still
 attempted. A cleanup failure on its own also fails the check.
@@ -72,3 +76,17 @@ PR updates cancel superseded runs. Rust dependency and pre-commit caches reduce
 repeated setup; caches do not substitute for tests or package validation. A cold
 cache must produce the same verdict as a warm cache. Failed tests are not
 automatically retried until they turn green.
+
+## Keep concurrency tests independent of machine speed
+
+Submitting a Turn or observing an Automation Run as `RUNNING` does not guarantee
+that the Agent has started consuming its input. Before interrupting a scripted
+Agent whose next step depends on consuming the current step, wait for an explicit
+signal from that Agent. Assert execution counts after the relevant work settles.
+Do not use a fixed sleep as evidence that work started or finished.
+
+The automation Goal-run suite runs every scenario with immediate and deferred
+Agent dispatch. The deferred variant deliberately delays Agent entry to expose
+assumptions about thread scheduling; it does not retry a failed test. Both variants
+must pass in each Python version. When a race is found, first reproduce the adverse
+ordering, then fix the synchronization and retain coverage for that ordering.

pyproject.toml

@@ -5,6 +5,12 @@ build-backend = "setuptools.build_meta"
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 pythonpath = ["."]
+# Emit thread stacks for a stalled test without changing its pass/fail budget.
+faulthandler_timeout = 60
+filterwarnings = [
+    "error::pytest.PytestUnhandledThreadExceptionWarning",
+    "error::pytest.PytestUnraisableExceptionWarning",
+]
 
 [tool.ruff]
 target-version = "py312"

tests/application/test_automation_goal_runs.py

@@ -131,6 +131,31 @@ def record_decision(self, index: int, decision: _Decision) -> None:
         with self._lock:
             self.decisions.append((index, decision))
 
+    def wait_for_started(self, prompt: str) -> None:
+        # A durable RUNNING Run does not mean its Agent claimed a script step.
+        _wait_until(
+            lambda: prompt in self.started_prompts,
+            f"Agent to consume its scripted step for {prompt!r}",
+        )
+
+
+@pytest.fixture(autouse=True, params=["immediate", "deferred"])
+def _agent_dispatch(request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch):
+    """Exercise both fast Agents and Agents that start after submission returns."""
+    if request.param == "immediate":
+        return
+
+    original = _GoalAwareSession.run_stream
+
+    async def deferred(self, op):
+        # Scheduling perturbation, not a readiness wait: assertions must still
+        # synchronize on the actual Agent or durable completion they inspect.
+        await asyncio.sleep(0.1)
+        async for event in original(self, op):
+            yield event
+
+    monkeypatch.setattr(_GoalAwareSession, "run_stream", deferred)
+
 
 def _application(
     tmp_path: Path,
@@ -447,15 +472,14 @@ def test_new_occurrence_during_active_run_is_terminal_skipped_without_turn(
             2,
             1,
         )
-        assert len(factory.started_prompts) == 1
-
         completion_gate.set()
         _wait_for_run(
             application,
             automation.id,
             active.run.id,
             AutomationRunStatus.COMPLETED,
         )
+        assert factory.started_prompts == [automation.prompt]
     finally:
         completion_gate.set()
         application.close()
@@ -699,6 +723,7 @@ def test_interrupted_turn_keeps_run_open_for_explicit_goal_continue(
         assert running.turn_id is not None
         assert running.goal_id is not None
 
+        factory.wait_for_started(automation.prompt)
         accepted, interrupted_turn = application.turns.interrupt(
             automation.thread_id,
             running.turn_id,
@@ -1041,13 +1066,7 @@ def inject_unreserved_turn(*args, **kwargs):
         assert foreign.prompt == "A legacy client won the race"
         assert foreign.id != execution.run.turn_id
 
-        # Submission can return before the Agent consumes its scripted step.
-        # Interrupt only after that step is claimed, so the Automation receives
-        # the second (completing) step rather than the foreign Turn's gate.
-        _wait_until(
-            lambda: factory.started_prompts == [foreign.prompt],
-            "foreign Agent to consume its scripted step",
-        )
+        factory.wait_for_started(foreign.prompt)
         accepted, interrupted = application.turns.interrupt(
             automation.thread_id,
             foreign.id,