test(regression): expand finite scenarios and share bounded execution lifecycle

tools/README.md

@@ -519,3 +519,72 @@ resource identifiers are not retained in this report. Unmatched requests are
 counted by method/status; webhook sub-paths are not separate declared routes.
 Missing logs or logs without recognizable access records are evidence errors.
 HTTP hits remain separate from assertion coverage.
+
+`regression_scenarios.expanded_profile()` builds 46 finite cases from the smoke
+fixtures (86 requests with two groups and three turns). It adds signed ingress
+rejection, API authentication, group reset/history isolation, tool recovery and
+empty retrieval cases. Save its returned profile alongside the existing fixture
+paths, bind `cli.config` to the target Workspace, and run it through
+`regression_suite.py`. Its three `mock-daily-*` models require the matching
+expanded mock scenario fixtures. The profile exercises one authenticated
+Workspace; OSS does not support creating a second Workspace. Multiple resources
+or group IDs do not establish cross-Workspace isolation.
+
+## Repeating finite scenarios within one budget
+
+Add this to a finite smoke/scenario profile to run its ordered cases repeatedly:
+
+```json
+"execution": {
+  "total_seconds": 1800,
+  "drain_seconds": 120,
+  "cooldown_seconds": 60
+}
+```
+
+Run it with the existing `regression_suite.py run --profile ... --output ...`
+entry point. Installation and plugin readiness precede the shared clock; the clock
+starts immediately before creating scenario resources. It covers preparation,
+case execution, owned-resource cleanup and final quiet sampling. The existing
+cleanup policy still applies; use `cleanup: "always"` in disposable deployments
+that must clean up after a failed run. Keep an outer process/environment supervisor
+as the hard wall-clock backstop: socket timeouts bound individual I/O waits, and
+local process termination cannot cancel work already accepted by the server.
+
+Every configured case and group turn is required in the first pass. A shortfall
+is `INCOMPLETE` and returns nonzero. Further cycles are optional: admission stops
+before the drain/cleanup reserve, including between cases and turns. An unfinished
+optional cycle does not fail the run by itself. Each attempted failure/unknown
+remains recorded; unknown results stop subsequent cycles after independent cases
+in the current cycle finish. Side-effecting requests are never retried. Known
+conversation AgentRuns are reconciled again at closeout for unexpected/unfinished
+runs. Outcomes without a session identity remain unknown.
+
+`cases.json` retains every attempted row and its cycle number; `cycles.json`
+records completion and status per cycle. `execution.json` records stage times.
+`resources.jsonl` uses the existing load/cooldown format plus the specific stage;
+existing `resource_limits` apply, including failure for missing recovery evidence.
+The final summary distinguishes `FAIL` and `INCOMPLETE`; both return nonzero.
+The CLI and deployment Job share the same resource cleanup, quiet sampling, final
+ledger reconciliation, coverage report and verdict. The Job preserves this
+sequence after partial resource preparation; its environment teardown follows
+the suite result and its own cleanup policy.
+Sampling starts before scenario resource creation, so choose the existing warmup
+window to cover the measured preparation and initial case warmup.
+
+Repeated cases create fresh group/conversation identities. The summary records
+observed session/run counts and the owned API resource count. History, Box reuse
+keys and idle expiry retain their product semantics, so repeated sessions can
+intentionally grow stored test data and live sandboxes. Resource budget failures
+remain failures; use an explicitly bounded profile when measuring steady reuse.
+Cleanup uses only `owned.json` and checks the authenticated Workspace. It does not
+delete arbitrary history, files, or containers; the deployment owner handles its
+outer teardown. The final quiet period lasts the configured cooldown, bounded by the remaining
+total budget. An early abort does not wait out the unused execution budget. Cleanup or drain exceeding its reserved endpoint is a failure.
+
+Finite scenarios record successful request rate and full-step P95 as observations
+(the step includes its verification). Load profiles can set
+`performance_mode: "observe"` to record throughput and latency while disabling
+only configured RPS/P95 thresholds. Business assertions, unresolved results,
+health failures, and process/container resource limits still determine success.
+The default load mode remains `gate` for existing benchmark users.

tools/regression_analysis.py

@@ -72,10 +72,11 @@ def container_metrics(row):
 def resource_verdict(directory, limits):
     validate_resource_limits(limits)
     rows = [json.loads(line) for line in (Path(directory) / 'resources.jsonl').read_text().splitlines()]
-    result = {'errors': [], 'targets': {}, 'limits': limits}
+    result = {'errors': [], 'incomplete': [], 'targets': {}, 'limits': limits}
     cooldown = [row for row in rows if row['phase'] == 'cooldown']
     if not cooldown:
-        result['errors'].append('No cooldown resource samples; recovery is unknown')
+        result['incomplete'].append('No cooldown resource samples; recovery is unknown')
+        result['errors'].extend(result['incomplete'])
         return result
     end = max(row['elapsed'] for row in cooldown)
     baseline = [
@@ -86,7 +87,8 @@ def resource_verdict(directory, limits):
     ]
     tail = [row for row in cooldown if row['elapsed'] >= end - limits['window_seconds']]
     if min(len(baseline), len(tail)) < limits['min_samples']:
-        result['errors'].append('Insufficient stable/cooldown samples; resource recovery is unknown')
+        result['incomplete'].append('Insufficient stable/cooldown samples; resource recovery is unknown')
+        result['errors'].extend(result['incomplete'])
         return result
     targets = dict(limits['targets'])
     if limits.get('containers'):
@@ -105,7 +107,7 @@ def resource_verdict(directory, limits):
                 not isinstance(v, (int, float)) or not math.isfinite(v) or v < 0 or (metric == 'rss_bytes' and v == 0)
                 for v in before + after
             ):
-                result['errors'].append(f'{name}/{metric}: missing or invalid samples')
+                result['incomplete'].append(f'{name}/{metric}: missing or invalid samples')
                 continue
             initial, final = statistics.median(before), statistics.median(after)
             growth = final - initial
@@ -117,6 +119,7 @@ def resource_verdict(directory, limits):
             }
             if growth > allowance:
                 result['errors'].append(f'{name}/{metric}: retained growth {growth:g} exceeds {allowance:g}')
+    result['errors'].extend(result['incomplete'])
     return result
 
 

tools/regression_contracts.json

@@ -5,169 +5,170 @@
     {"id": "configuration.identity",
      "area": "configuration", "kind": "blackbox", "priority": "P1",
      "description": "CLI resolves the configured Workspace identity",
-     "bindings": [{"profile": "smoke", "case": "identity", "checks": ["response"]}]
+     "bindings": [{"profile": "smoke", "case": "identity", "checks": ["response"]}, {"profile": "scenarios", "case": "identity", "checks": ["response"]}]
     },
     {"id": "configuration.plugins",
      "area": "configuration", "kind": "blackbox", "priority": "P1",
      "description": "Installed LocalAgent is visible through the plugin API",
-     "bindings": [{"profile": "smoke", "case": "plugins", "checks": ["response"]}],
+     "bindings": [{"profile": "smoke", "case": "plugins", "checks": ["response"]}, {"profile": "scenarios", "case": "plugins", "checks": ["response"]}],
      "routes": ["GET /api/v1/plugins"]
     },
     {"id": "configuration.provider_readback",
      "area": "configuration", "kind": "blackbox", "priority": "P1",
      "description": "Created provider is visible through the public API",
-     "bindings": [{"profile": "smoke", "case": "provider_readback", "checks": ["response"]}],
+     "bindings": [{"profile": "smoke", "case": "provider_readback", "checks": ["response"]}, {"profile": "scenarios", "case": "provider_readback", "checks": ["response"]}],
      "routes": ["GET /api/v1/provider/providers/<provider_uuid>"]
     },
     {"id": "configuration.model_readback",
      "area": "configuration", "kind": "blackbox", "priority": "P1",
      "description": "Created model is visible through the public API",
-     "bindings": [{"profile": "smoke", "case": "model_readback", "checks": ["response"]}],
+     "bindings": [{"profile": "smoke", "case": "model_readback", "checks": ["response"]}, {"profile": "scenarios", "case": "model_readback", "checks": ["response"]}],
      "routes": ["GET /api/v1/provider/models/llm/<model_uuid>"]
     },
     {"id": "configuration.pipeline_readback",
      "area": "configuration", "kind": "blackbox", "priority": "P1",
      "description": "Pipeline readback includes its configured model",
-     "bindings": [{"profile": "smoke", "case": "pipeline_readback", "checks": ["response"]}],
+     "bindings": [{"profile": "smoke", "case": "pipeline_readback", "checks": ["response"]}, {"profile": "scenarios", "case": "pipeline_readback", "checks": ["response"]}],
      "routes": ["GET /api/v1/pipelines/<pipeline_uuid>"]
     },
     {"id": "configuration.kb_readback",
      "area": "configuration", "kind": "blackbox", "priority": "P1",
      "description": "Knowledge-base identity and engine match configuration",
-     "bindings": [{"profile": "smoke", "case": "kb_readback", "checks": ["response", "fields"]}],
+     "bindings": [{"profile": "smoke", "case": "kb_readback", "checks": ["response", "fields"]}, {"profile": "scenarios", "case": "kb_readback", "checks": ["response", "fields"]}],
      "routes": ["GET /api/v1/knowledge/bases/<knowledge_base_uuid>"]
     },
     {"id": "configuration.document_readback",
      "area": "configuration", "kind": "blackbox", "priority": "P1",
      "description": "Uploaded document completes ingestion in its owning knowledge base",
-     "bindings": [{"profile": "smoke", "case": "document_readback", "checks": ["response", "fields"]}],
+     "bindings": [{"profile": "smoke", "case": "document_readback", "checks": ["response", "fields"]}, {"profile": "scenarios", "case": "document_readback", "checks": ["response", "fields"]}],
      "routes": ["GET /api/v1/knowledge/bases/<knowledge_base_uuid>/files"]
     },
     {"id": "ingress.missing_signature",
      "area": "ingress", "kind": "blackbox", "priority": "P0",
      "description": "Unsigned bot messages are rejected",
-     "bindings": [{"profile": "smoke", "case": "missing_signature", "checks": ["response", "fields"]}]
+     "bindings": [{"profile": "smoke", "case": "missing_signature", "checks": ["response", "fields"]}, {"profile": "scenarios", "case": "missing_signature", "checks": ["response", "fields"]}]
     },
     {"id": "ingress.bad_signature",
      "area": "ingress", "kind": "blackbox", "priority": "P0",
      "description": "Messages with an invalid signature are rejected",
-     "bindings": [{"profile": "smoke", "case": "bad_signature", "checks": ["response", "fields"]}]
+     "bindings": [{"profile": "smoke", "case": "bad_signature", "checks": ["response", "fields"]}, {"profile": "scenarios", "case": "bad_signature", "checks": ["response", "fields"]}]
     },
     {"id": "conversation.chat",
      "area": "conversation", "kind": "blackbox", "priority": "P1",
      "description": "Pipeline returns the expected chat response",
-     "bindings": [{"profile": "smoke", "case": "chat", "checks": ["response"]}, {"profile": "endurance", "case": "chat", "checks": ["response"]}],
+     "bindings": [{"profile": "smoke", "case": "chat", "checks": ["response"]}, {"profile": "endurance", "case": "chat", "checks": ["response"]}, {"profile": "scenarios", "case": "chat", "checks": ["response"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "conversation.unicode",
      "area": "conversation", "kind": "blackbox", "priority": "P1",
      "description": "Unicode input and streamed output survive the full conversation chain",
-     "bindings": [{"profile": "smoke", "case": "unicode", "checks": ["response"]}],
+     "bindings": [{"profile": "smoke", "case": "unicode", "checks": ["response"]}, {"profile": "scenarios", "case": "unicode", "checks": ["response"]}, {"profile": "scenarios", "case": "group_unicode", "checks": ["response"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "model.slow",
      "area": "model", "kind": "blackbox", "priority": "P1",
      "description": "Slow streamed model output reaches the caller",
-     "bindings": [{"profile": "smoke", "case": "slow", "checks": ["response", "model_sequence"]}],
+     "bindings": [{"profile": "smoke", "case": "slow", "checks": ["response", "model_sequence"]}, {"profile": "scenarios", "case": "slow", "checks": ["response", "model_sequence"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "model.fallback401",
      "area": "model", "kind": "blackbox", "priority": "P1",
      "description": "Authentication failure falls back to the configured secondary model",
-     "bindings": [{"profile": "smoke", "case": "fallback401", "checks": ["response", "model_sequence"]}],
+     "bindings": [{"profile": "smoke", "case": "fallback401", "checks": ["response", "model_sequence"]}, {"profile": "scenarios", "case": "fallback401", "checks": ["response", "model_sequence"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "model.fallback429",
      "area": "model", "kind": "blackbox", "priority": "P1",
      "description": "Rate limiting falls back to the configured secondary model",
-     "bindings": [{"profile": "smoke", "case": "fallback429", "checks": ["response", "model_sequence"]}],
+     "bindings": [{"profile": "smoke", "case": "fallback429", "checks": ["response", "model_sequence"]}, {"profile": "scenarios", "case": "fallback429", "checks": ["response", "model_sequence"]}, {"profile": "scenarios", "case": "group_fallback429", "checks": ["response", "model_sequence"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "model.fallback503",
      "area": "model", "kind": "blackbox", "priority": "P1",
      "description": "Unavailable primary model falls back to the secondary model",
-     "bindings": [{"profile": "smoke", "case": "fallback503", "checks": ["response", "model_sequence"]}],
+     "bindings": [{"profile": "smoke", "case": "fallback503", "checks": ["response", "model_sequence"]}, {"profile": "scenarios", "case": "fallback503", "checks": ["response", "model_sequence"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "tools.exec_error",
      "area": "tools", "kind": "blackbox", "priority": "P1",
      "description": "Tool error output is delivered to the model and handled",
-     "bindings": [{"profile": "smoke", "case": "exec_error", "checks": ["response"]}],
+     "bindings": [{"profile": "smoke", "case": "exec_error", "checks": ["response"]}, {"profile": "scenarios", "case": "exec_error", "checks": ["response"]}, {"profile": "scenarios", "case": "group_exec_error", "checks": ["response"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "tools.parallel",
      "area": "tools", "kind": "blackbox", "priority": "P1",
      "description": "Parallel tool calls return matched results to the model",
-     "bindings": [{"profile": "smoke", "case": "parallel", "checks": ["response"]}],
+     "bindings": [{"profile": "smoke", "case": "parallel", "checks": ["response"]}, {"profile": "scenarios", "case": "parallel", "checks": ["response"]}, {"profile": "scenarios", "case": "group_parallel", "checks": ["response"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "delivery.unicode_export",
      "area": "delivery", "kind": "blackbox", "priority": "P1",
      "description": "Exported Unicode file name and bytes survive delivery",
-     "bindings": [{"profile": "smoke", "case": "unicode_export", "checks": ["response", "attachment"]}],
+     "bindings": [{"profile": "smoke", "case": "unicode_export", "checks": ["response", "attachment"]}, {"profile": "scenarios", "case": "unicode_export", "checks": ["response", "attachment"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "retrieval.other_rag",
      "area": "retrieval", "kind": "blackbox", "priority": "P1",
      "description": "A second knowledge base supplies its own content to the model",
-     "bindings": [{"profile": "smoke", "case": "other_rag", "checks": ["response"]}],
+     "bindings": [{"profile": "smoke", "case": "other_rag", "checks": ["response"]}, {"profile": "scenarios", "case": "other_rag", "checks": ["response"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "tools.box",
      "area": "tools", "kind": "blackbox", "priority": "P1",
      "description": "A real Box write/read round trip reaches the final reply",
-     "bindings": [{"profile": "smoke", "case": "box", "checks": ["response"]}, {"profile": "endurance", "case": "box", "checks": ["response"]}],
+     "bindings": [{"profile": "smoke", "case": "box", "checks": ["response"]}, {"profile": "endurance", "case": "box", "checks": ["response"]}, {"profile": "scenarios", "case": "box", "checks": ["response"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "delivery.box_export",
      "area": "delivery", "kind": "blackbox", "priority": "P1",
      "description": "Exported file bytes match the generated fixture",
-     "bindings": [{"profile": "smoke", "case": "box_export", "checks": ["response", "attachment"]}],
+     "bindings": [{"profile": "smoke", "case": "box_export", "checks": ["response", "attachment"]}, {"profile": "scenarios", "case": "box_export", "checks": ["response", "attachment"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "retrieval.rag_chat",
      "area": "retrieval", "kind": "blackbox", "priority": "P1",
      "description": "Retrieved fixture content reaches the model and final reply",
-     "bindings": [{"profile": "smoke", "case": "rag_chat", "checks": ["response"]}, {"profile": "endurance", "case": "rag_chat", "checks": ["response"]}],
+     "bindings": [{"profile": "smoke", "case": "rag_chat", "checks": ["response"]}, {"profile": "endurance", "case": "rag_chat", "checks": ["response"]}, {"profile": "scenarios", "case": "rag_chat", "checks": ["response"]}],
      "routes": ["POST /api/v1/pipelines/<pipeline_uuid>/run"]
     },
     {"id": "retrieval.rag_retrieve",
      "area": "retrieval", "kind": "blackbox", "priority": "P0",
      "description": "Retrieval returns the owning knowledge-base marker without the other fixture",
-     "bindings": [{"profile": "smoke", "case": "rag_retrieve", "checks": ["response", "fields", "isolation"]}],
+     "bindings": [{"profile": "smoke", "case": "rag_retrieve", "checks": ["response", "fields", "isolation"]}, {"profile": "scenarios", "case": "rag_retrieve", "checks": ["response", "fields", "isolation"]}],
      "routes": ["POST /api/v1/knowledge/bases/<knowledge_base_uuid>/retrieve"]
     },
     {"id": "retrieval.retrieve_other",
      "area": "retrieval", "kind": "blackbox", "priority": "P0",
      "description": "Retrieval from the second knowledge base excludes the first fixture",
-     "bindings": [{"profile": "smoke", "case": "retrieve_other", "checks": ["response", "fields", "isolation"]}],
+     "bindings": [{"profile": "smoke", "case": "retrieve_other", "checks": ["response", "fields", "isolation"]}, {"profile": "scenarios", "case": "retrieve_other", "checks": ["response", "fields", "isolation"]}],
      "routes": ["POST /api/v1/knowledge/bases/<knowledge_base_uuid>/retrieve"]
     },
     {"id": "retrieval.retrieve_empty",
      "area": "retrieval", "kind": "blackbox", "priority": "P1",
      "description": "Empty knowledge base returns an empty retrieval result",
-     "bindings": [{"profile": "smoke", "case": "retrieve_empty", "checks": ["response", "fields"]}],
+     "bindings": [{"profile": "smoke", "case": "retrieve_empty", "checks": ["response", "fields"]}, {"profile": "scenarios", "case": "retrieve_empty", "checks": ["response", "fields"]}],
      "routes": ["POST /api/v1/knowledge/bases/<knowledge_base_uuid>/retrieve"]
     },
     {"id": "conversation.group_chat_history",
      "area": "conversation", "kind": "blackbox", "priority": "P0",
      "description": "Multi-turn chat preserves the prior reply without mixing group history",
-     "bindings": [{"profile": "smoke", "case": "group_chat_history", "checks": ["response"], "min_turn": 2}, {"profile": "endurance", "case": "chat", "checks": ["response"], "min_turn": 2}]
+     "bindings": [{"profile": "smoke", "case": "group_chat_history", "checks": ["response"], "min_turn": 2}, {"profile": "endurance", "case": "chat", "checks": ["response"], "min_turn": 2}, {"profile": "scenarios", "case": "group_chat_history", "checks": ["response"], "min_turn": 2}]
     },
     {"id": "conversation.group_box_history",
      "area": "conversation", "kind": "blackbox", "priority": "P1",
      "description": "Multi-turn Box conversation preserves group history",
-     "bindings": [{"profile": "smoke", "case": "group_box_history", "checks": ["response"], "min_turn": 2}, {"profile": "endurance", "case": "box", "checks": ["response"], "min_turn": 2}]
+     "bindings": [{"profile": "smoke", "case": "group_box_history", "checks": ["response"], "min_turn": 2}, {"profile": "endurance", "case": "box", "checks": ["response"], "min_turn": 2}, {"profile": "scenarios", "case": "group_box_history", "checks": ["response"], "min_turn": 2}]
     },
     {"id": "conversation.group_rag_history",
      "area": "conversation", "kind": "blackbox", "priority": "P1",
      "description": "Multi-turn RAG conversation preserves group history",
-     "bindings": [{"profile": "smoke", "case": "group_rag_history", "checks": ["response"], "min_turn": 2}, {"profile": "endurance", "case": "rag_chat", "checks": ["response"], "min_turn": 2}]
+     "bindings": [{"profile": "smoke", "case": "group_rag_history", "checks": ["response"], "min_turn": 2}, {"profile": "endurance", "case": "rag_chat", "checks": ["response"], "min_turn": 2}, {"profile": "scenarios", "case": "group_rag_history", "checks": ["response"], "min_turn": 2}]
     },
     {"id": "auth.invalid_credentials",
      "area": "auth", "kind": "blackbox", "priority": "P0",
-     "description": "Missing and invalid API credentials cannot access Workspace resources",
-     "bindings": []
+     "description": "Invalid API credentials cannot access Workspace resources",
+     "bindings": [{"profile": "scenarios", "case": "api_auth_invalid", "checks": ["response", "fields"]}],
+     "routes": ["GET /api/v1/pipelines"]
     },
     {"id": "auth.revocation",
      "area": "auth", "kind": "blackbox", "priority": "P0",
@@ -227,7 +228,7 @@
     {"id": "tools.timeout_recovery",
      "area": "tools", "kind": "blackbox", "priority": "P1",
      "description": "Timed-out tools stop running and the next tool can execute",
-     "bindings": []
+     "bindings": [{"profile": "scenarios", "case": "tool_timeout_recovery", "checks": ["response"]}]
     },
     {"id": "delivery.callback_retry",
      "area": "delivery", "kind": "blackbox", "priority": "P1",
@@ -267,17 +268,103 @@
     {"id": "execution.persisted_terminal",
      "area": "execution", "kind": "graybox", "priority": "P0",
      "description": "Each executed request has a unique completed persisted run and terminal event",
-     "bindings": [{"profile": "smoke", "case": "chat", "checks": ["ledger"]}, {"profile": "endurance", "case": "chat", "checks": ["ledger"]}]
+     "bindings": [{"profile": "smoke", "case": "chat", "checks": ["ledger"]}, {"profile": "endurance", "case": "chat", "checks": ["ledger"]}, {"profile": "scenarios", "case": "chat", "checks": ["ledger"]}]
     },
     {"id": "tools.persisted_box_results",
      "area": "tools", "kind": "graybox", "priority": "P0",
      "description": "Persisted Box events match backend, exit status and output",
-     "bindings": [{"profile": "smoke", "case": "box", "checks": ["ledger"]}, {"profile": "endurance", "case": "box", "checks": ["ledger"]}]
+     "bindings": [{"profile": "smoke", "case": "box", "checks": ["ledger"]}, {"profile": "endurance", "case": "box", "checks": ["ledger"]}, {"profile": "scenarios", "case": "box", "checks": ["ledger"]}]
     },
     {"id": "ingress.no_run_on_reject",
      "area": "ingress", "kind": "graybox", "priority": "P0",
      "description": "Rejected ingress does not create an AgentRun",
-     "bindings": [{"profile": "smoke", "case": "missing_signature", "checks": ["no_run"]}]
+     "bindings": [{"profile": "smoke", "case": "missing_signature", "checks": ["no_run"]}, {"profile": "scenarios", "case": "missing_signature", "checks": ["no_run"]}, {"profile": "scenarios", "case": "expired_signature", "checks": ["no_run"]}, {"profile": "scenarios", "case": "future_signature", "checks": ["no_run"]}, {"profile": "scenarios", "case": "bad_timestamp", "checks": ["no_run"]}, {"profile": "scenarios", "case": "tampered_body", "checks": ["no_run"]}, {"profile": "scenarios", "case": "invalid_json", "checks": ["no_run"]}, {"profile": "scenarios", "case": "invalid_object", "checks": ["no_run"]}, {"profile": "scenarios", "case": "missing_session", "checks": ["no_run"]}, {"profile": "scenarios", "case": "invalid_message", "checks": ["no_run"]}]
+    },
+    {"id": "ingress.expired_signature",
+     "area": "ingress", "kind": "blackbox", "priority": "P0",
+     "description": "Expired signed messages are rejected",
+     "bindings": [{"profile": "scenarios", "case": "expired_signature", "checks": ["response", "fields"]}]
+    },
+    {"id": "ingress.future_signature",
+     "area": "ingress", "kind": "blackbox", "priority": "P0",
+     "description": "Signed messages too far in the future are rejected",
+     "bindings": [{"profile": "scenarios", "case": "future_signature", "checks": ["response", "fields"]}]
+    },
+    {"id": "ingress.bad_timestamp",
+     "area": "ingress", "kind": "blackbox", "priority": "P0",
+     "description": "Malformed signing timestamps are rejected",
+     "bindings": [{"profile": "scenarios", "case": "bad_timestamp", "checks": ["response", "fields"]}]
+    },
+    {"id": "ingress.tampered_body",
+     "area": "ingress", "kind": "blackbox", "priority": "P0",
+     "description": "A changed body cannot reuse a valid signature",
+     "bindings": [{"profile": "scenarios", "case": "tampered_body", "checks": ["response", "fields"]}]
+    },
+    {"id": "ingress.invalid_json",
+     "area": "ingress", "kind": "blackbox", "priority": "P0",
+     "description": "Validly signed malformed JSON is rejected",
+     "bindings": [{"profile": "scenarios", "case": "invalid_json", "checks": ["response", "fields"]}]
+    },
+    {"id": "ingress.invalid_object",
+     "area": "ingress", "kind": "blackbox", "priority": "P0",
+     "description": "Validly signed non-object JSON is rejected",
+     "bindings": [{"profile": "scenarios", "case": "invalid_object", "checks": ["response", "fields"]}]
+    },
+    {"id": "ingress.missing_session",
+     "area": "ingress", "kind": "blackbox", "priority": "P0",
+     "description": "Messages without a session identity are rejected",
+     "bindings": [{"profile": "scenarios", "case": "missing_session", "checks": ["response", "fields"]}]
+    },
+    {"id": "ingress.invalid_message",
+     "area": "ingress", "kind": "blackbox", "priority": "P0",
+     "description": "Invalid message chains are rejected",
+     "bindings": [{"profile": "scenarios", "case": "invalid_message", "checks": ["response", "fields"]}]
+    },
+    {"id": "auth.missing_credentials",
+     "area": "auth", "kind": "blackbox", "priority": "P0",
+     "description": "Unauthenticated requests cannot access Workspace resources",
+     "bindings": [{"profile": "scenarios", "case": "api_auth_missing", "checks": ["response", "fields"]}],
+     "routes": ["GET /api/v1/pipelines"]
+    },
+    {"id": "conversation.group_unicode",
+     "area": "conversation", "kind": "blackbox", "priority": "P1",
+     "description": "Multiple group turns preserve history with Unicode replies",
+     "bindings": [{"profile": "scenarios", "case": "group_unicode", "checks": ["response"], "min_turn": 2}]
+    },
+    {"id": "conversation.group_fallback429",
+     "area": "conversation", "kind": "blackbox", "priority": "P1",
+     "description": "Multiple group turns preserve history with model fallback",
+     "bindings": [{"profile": "scenarios", "case": "group_fallback429", "checks": ["response"], "min_turn": 2}]
+    },
+    {"id": "conversation.group_exec_error",
+     "area": "conversation", "kind": "blackbox", "priority": "P1",
+     "description": "Multiple group turns preserve history with handled tool errors",
+     "bindings": [{"profile": "scenarios", "case": "group_exec_error", "checks": ["response"], "min_turn": 2}]
+    },
+    {"id": "conversation.group_parallel",
+     "area": "conversation", "kind": "blackbox", "priority": "P1",
+     "description": "Multiple group turns preserve history with parallel tools",
+     "bindings": [{"profile": "scenarios", "case": "group_parallel", "checks": ["response"], "min_turn": 2}]
+    },
+    {"id": "conversation.reset_history",
+     "area": "conversation", "kind": "blackbox", "priority": "P0",
+     "description": "Reset clears the selected group history before the next turn",
+     "bindings": [{"profile": "scenarios", "case": "group_reset_isolation", "checks": ["reset", "response"], "group_index": 0, "turn": 2}]
+    },
+    {"id": "conversation.reset_isolation",
+     "area": "conversation", "kind": "blackbox", "priority": "P0",
+     "description": "Resetting one group preserves the other group history",
+     "bindings": [{"profile": "scenarios", "case": "group_reset_isolation", "checks": ["response"], "group_index": 1, "min_turn": 2}]
+    },
+    {"id": "tools.failure_then_success",
+     "area": "tools", "kind": "blackbox", "priority": "P1",
+     "description": "A failed tool is followed by a successful tool in the same run",
+     "bindings": [{"profile": "scenarios", "case": "tool_recovery", "checks": ["response"]}]
+    },
+    {"id": "retrieval.empty_context_chat",
+     "area": "retrieval", "kind": "blackbox", "priority": "P1",
+     "description": "An empty knowledge base allows chat without injecting other knowledge fixtures",
+     "bindings": [{"profile": "scenarios", "case": "empty_rag_chat", "checks": ["response"]}]
     }
   ]
 }

tools/regression_coverage.py

@@ -228,6 +228,10 @@ def report(catalog, rows, errors=()):
             for row in rows:
                 if (row['profile'], row['case']) != (binding['profile'], binding['case']):
                     continue
+                if 'group_index' in binding and row.get('group_index') != binding['group_index']:
+                    continue
+                if 'turn' in binding and row.get('turn') != binding['turn']:
+                    continue
                 if row.get('turn', 0) < binding.get('min_turn', 0):
                     continue
                 checks = [row.get('assertions', {}).get(name) for name in binding['checks']]

tools/regression_job.py

@@ -102,6 +102,7 @@ class Job:
         self.env = {**os.environ, 'XDG_CONFIG_HOME': str(self.private / 'config')}
         self.cli = [profile.get('lbctl', 'lbctl'), '--config', str(self.cli_config), '-o', 'json']
         self.redactions = []
+        self.suite = None
         self.state = {'job_id': secrets.token_hex(8), 'record': None, 'bound': False, 'phase': 'created'}
 
     def redact(self, value):
@@ -347,13 +348,14 @@ class Job:
             pid = int(self.checked(['docker', 'inspect', '--format', '{{.State.Pid}}', container]).strip())
             name = (
                 {'langbot': 'core', 'langbot_plugin_runtime': 'runtime', 'langbot_box': 'box'}.get(service, service)
-                if profile.get('verification')
+                if profile.get('verification') or profile.get('execution')
                 else service
             )
             profile['observe']['pids'][name] = pid
+        profile['observe']['box_data_root'] = str(Path(self.state['record']['data_dir']) / 'box')
         destination = self.output / 'suite'
         destination.mkdir(mode=0o700)
-        suite = Suite(profile, self.suite_path, destination, env=self.env)
+        suite = self.suite = Suite(profile, self.suite_path, destination, env=self.env)
         self.redactions.extend(suite.secrets)
         if published is not None:
             self.state['phase'] = 'plugins'
@@ -377,7 +379,6 @@ class Job:
             data = suite.call(case['args'], case.get('body'))
             if case['expect'] not in json.dumps(data, ensure_ascii=False):
                 raise RuntimeError(f'warmup oracle failed: {case["name"]}')
-        profile['observe']['box_data_root'] = str(Path(self.state['record']['data_dir']) / 'box')
         suite.write('profile.json', profile)
         return suite
 
@@ -436,14 +437,11 @@ class Job:
         self.save()
         result = {'errors': [], 'completed': False}
         try:
-            suite = self.prepare()
+            suite = self.suite = self.prepare()
             self.state['phase'] = 'traffic'
             self.save()
             summary = suite.run()
-            suite.finish(summary)
             result.update(completed=True, suite=summary, errors=list(summary['errors']))
-            self.state['phase'] = 'finished'
-            self.save()
         except KeyboardInterrupt:
             result['errors'].append('Job interrupted; in-flight outcomes may be unknown')
         except (RuntimeError, ValueError, OSError, KeyError, subprocess.TimeoutExpired) as exc:
@@ -451,6 +449,17 @@ class Job:
         finally:
             if not result['completed'] and not result['errors']:
                 result['errors'].append('Job aborted before completion')
+            if self.suite is not None:
+                summary = result.setdefault('suite', {'phases': [], 'errors': list(result['errors'])})
+                try:
+                    self.suite.finish(summary)
+                except (Exception, KeyboardInterrupt) as exc:
+                    summary['errors'].append(f'suite finalization: {type(exc).__name__}: ' + self.redact(str(exc)))
+                    summary.update(passed=False, outcome='FAIL')
+                result['errors'] = list(dict.fromkeys([*result['errors'], *summary['errors']]))
+            if result['completed']:
+                self.state['phase'] = 'finished'
+                self.save()
             try:
                 self.collect()
             except (RuntimeError, ValueError, OSError, subprocess.TimeoutExpired) as exc:

tools/regression_scenarios.py

@@ -0,0 +1,240 @@
+"""Expanded finite daily cases built from the maintained PR smoke fixtures."""
+
+import copy
+import json
+import subprocess
+import threading
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).parent
+RUNNER = 'plugin:langbot-team/LocalAgent/default'
+
+
+def expanded_profile():
+    profile = json.loads((ROOT / 'regression_profiles/smoke.json').read_text())
+    profile['coverage_suite'] = 'scenarios'
+    profile['smoke']['group_turns'] = 3
+    resources = {r['id']: r for r in profile['resources']}
+    cases = {c['name']: c for c in profile['cases']}
+
+    for rejection in (
+        'expired_signature',
+        'future_signature',
+        'bad_timestamp',
+        'tampered_body',
+        'invalid_json',
+        'invalid_object',
+        'missing_session',
+        'invalid_message',
+    ):
+        case = copy.deepcopy(cases['missing_signature'])
+        case.update(name=rejection, reject_group=rejection)
+        if rejection in ('invalid_json', 'invalid_object', 'missing_session', 'invalid_message'):
+            case.update(expect='40001', fields={'http_status': 400, 'code': 40001})
+        profile['cases'].append(case)
+
+    for auth in ('missing', 'invalid'):
+        profile['cases'].append(
+            {
+                'name': f'api_auth_{auth}',
+                'stage': 'auth',
+                'http_probe': {'url': '${core_base_url}/api/v1/pipelines', 'auth': auth},
+                'expect': 'http_status',
+                'fields': {'http_status': 401},
+            }
+        )
+
+    def group_case(name, pipeline, message=None):
+        bot = copy.deepcopy(resources['chat_bot'])
+        bot['id'] = name + '_bot'
+        bot['body']['name'] = 'daily ' + name
+        bot['body']['event_bindings'][0]['target_uuid'] = '${' + pipeline + '}'
+        profile['resources'].append(bot)
+        case = copy.deepcopy(cases['group_chat_history'])
+        case.update(name=name)
+        case['http_group']['url'] = '${core_base_url}/bots/${' + bot['id'] + '}/sync'
+        if message:
+            case['http_group']['message'] = message
+        return case
+
+    for source, message in [
+        ('unicode', 'ai 请原样处理：你好，世界 🌍'),
+        ('fallback429', None),
+        ('exec_error', None),
+        ('parallel', None),
+    ]:
+        case = group_case('group_' + source, source + '_pipeline', message)
+        for key in ('expect', 'box_exec', 'mock_requests'):
+            if key in cases[source]:
+                case[key] = copy.deepcopy(cases[source][key])
+        profile['cases'].append(case)
+
+    reset = copy.deepcopy(cases['group_chat_history'])
+    reset.update(name='group_reset_isolation', reset_group=True)
+    profile['cases'].append(reset)
+
+    for name in ('tool_recovery', 'tool_timeout_recovery', 'empty_rag_chat'):
+        model = copy.deepcopy(resources['chat_model'])
+        model['id'] = name + '_model'
+        model['body']['name'] = 'mock-daily-' + name.replace('_', '-')
+        pipeline = copy.deepcopy(resources['box_pipeline' if name.startswith('tool_') else 'rag_pipeline'])
+        pipeline['id'] = name + '_pipeline'
+        pipeline['body']['name'] = 'daily ' + name
+        config = pipeline['body']['config']['ai']['runner_config'][RUNNER]
+        config['model'] = {'primary': '${' + model['id'] + '}', 'fallbacks': []}
+        if name == 'empty_rag_chat':
+            config['knowledge-bases'] = ['${empty_knowledge_base}']
+        profile['resources'].extend([model, pipeline])
+        case = copy.deepcopy(cases['chat'])
+        case.update(
+            name=name,
+            stage='tools' if name.startswith('tool_') else 'retrieval',
+            args=[
+                'pipeline',
+                'run',
+                '${' + pipeline['id'] + '}',
+                '--message',
+                'Run the daily scenario',
+                '--timeout',
+                '90s',
+            ],
+            expect='LANGBOT_' + name.upper() + '_OK',
+        )
+        profile['cases'].append(case)
+    return profile
+
+
+class ExecutionWindow:
+    """One shared scenario/cleanup deadline, with the existing resource sampler format."""
+
+    def __init__(self, suite):
+        self.suite = suite
+        self.settings = suite.profile['execution']
+        self.started = time.monotonic()
+        self.deadline = self.started + self.settings['total_seconds']
+        self.cleanup_deadline = self.deadline - self.settings['cooldown_seconds']
+        self.admission_deadline = self.cleanup_deadline - self.settings['drain_seconds']
+        self.stage = 'prepare'
+        self.phases = []
+        self.errors = []
+        self.stop = threading.Event()
+        self.thread = threading.Thread(target=self.sample, daemon=True)
+
+    def start(self):
+        self.mark('prepare')
+        self.thread.start()
+
+    def mark(self, stage):
+        self.stage = stage
+        self.phases.append({'stage': stage, 'time': time.time(), 'elapsed': time.monotonic() - self.started})
+        self.suite.write('execution.json', {'settings': self.settings, 'phases': self.phases})
+
+    def timeout(self, maximum, *, sampling=False):
+        remaining = (self.deadline if sampling else self.cleanup_deadline) - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError('Shared execution deadline reached; no operation started')
+        return min(maximum, remaining)
+
+    def admitting(self):
+        return time.monotonic() < self.admission_deadline
+
+    def sample(self):
+        from regression_suite import resolve
+        from regression_traffic import (
+            owned_box_containers,
+            read_endpoint_snapshot,
+            read_process_snapshot,
+            sample_containers,
+        )
+
+        def run(command, **kwargs):
+            kwargs['timeout'] = self.timeout(kwargs.get('timeout', 10), sampling=True)
+            return subprocess.run(command, **kwargs)
+
+        def discover(root):
+            def checked(command):
+                result = run(command, capture_output=True, text=True, check=True)
+                return result.stdout
+
+            return owned_box_containers(root, checked=checked)
+
+        try:
+            observed = resolve(self.suite.profile['observe'], self.suite.variables)
+            pids = dict(observed.get('pids', {}))
+            for name, path in observed.get('pid_files', {}).items():
+                pids[name] = int(Path(self.suite.path(path)).read_text().strip())
+            with (self.suite.output / 'resources.jsonl').open('w') as stream:
+                while not self.stop.is_set() and time.monotonic() < self.deadline:
+                    stage = self.stage
+                    row = {
+                        'time': time.time(),
+                        'elapsed': time.monotonic() - self.started,
+                        'phase': 'cooldown' if stage == 'cooldown' else 'load',
+                        'stage': stage,
+                        'targets': {},
+                        'health': {},
+                    }
+                    for name, pid in pids.items():
+                        try:
+                            row['targets'][name] = read_process_snapshot(pid)
+                        except (OSError, RuntimeError) as exc:
+                            row['targets'][name] = {'error': str(exc)}
+                            self.errors.append(f'{name}: process sample failed: {exc}')
+                    for name, url in observed['health'].items():
+                        try:
+                            row['health'][name] = read_endpoint_snapshot(
+                                url, timeout_seconds=self.timeout(3, sampling=True)
+                            )
+                        except (OSError, RuntimeError, ValueError) as exc:
+                            row['health'][name] = {'error': str(exc)}
+                            self.errors.append(f'{name}: health sample failed: {exc}')
+                    if observed.get('containers') or observed.get('box_data_root'):
+                        row.update(
+                            sample_containers(
+                                observed.get('containers', []),
+                                observed.get('box_data_root'),
+                                run=run,
+                                discover=discover,
+                            )
+                        )
+                        if row.get('container_error'):
+                            self.errors.append('Container resource sample failed')
+                    stream.write(json.dumps(row) + '\n')
+                    stream.flush()
+                    self.stop.wait(3)
+        except (OSError, RuntimeError, ValueError, KeyError, subprocess.SubprocessError) as exc:
+            self.errors.append(f'Resource sampler stopped: {exc}')
+
+    def finish(self, result):
+        if time.monotonic() > self.cleanup_deadline:
+            result['errors'].append('Execution/drain/cleanup exceeded the shared deadline')
+        self.mark('cooldown')
+        remaining = max(0, self.deadline - time.monotonic() - min(1, self.settings['cooldown_seconds'] / 10))
+        self.stop.wait(min(self.settings['cooldown_seconds'], remaining))
+        self.stop.set()
+        self.thread.join(timeout=max(0, self.deadline - time.monotonic()))
+        if self.thread.is_alive():
+            result['errors'].append('Resource sampler did not stop before its drain deadline')
+        result['errors'].extend(dict.fromkeys(self.errors))
+        if limits := self.suite.profile.get('resource_limits'):
+            from regression_analysis import resource_verdict
+
+            try:
+                verdict = resource_verdict(self.suite.output, limits)
+                result['resources'] = verdict
+                result['errors'].extend(verdict['errors'])
+                result.setdefault('incomplete', []).extend(verdict['incomplete'])
+            except (OSError, ValueError, KeyError) as exc:
+                error = f'Resource recovery evidence unavailable: {exc}'
+                result['errors'].append(error)
+                result.setdefault('incomplete', []).append(error)
+        from regression_smoke import reconcile_scenario_runs
+
+        reconcile_scenario_runs(self.suite, result, key='ledger_closeout')
+        self.mark('finished')
+        result['execution'] = {
+            'settings': self.settings,
+            'phases': self.phases,
+            'elapsed_seconds': time.monotonic() - self.started,
+        }

tools/regression_smoke.py

@@ -4,6 +4,9 @@ from collections import Counter
 from contextlib import closing
 import base64
 import json
+import hashlib
+import hmac
+import os
 from pathlib import Path
 import sqlite3
 import time
@@ -11,7 +14,7 @@ import urllib.error
 import urllib.request
 from uuid import uuid4
 
-from regression_traffic import group_request, read_endpoint_snapshot
+from regression_traffic import group_request, percentile, read_endpoint_snapshot, reconcile_runs
 from regression_coverage import checked, check_response
 
 
@@ -104,8 +107,8 @@ def check_fields(data, expected):
     return expected
 
 
-def check_mock_requests(path, offset, expected):
-    deadline = time.monotonic() + 3
+def check_mock_requests(path, offset, expected, *, timeout=3):
+    deadline = time.monotonic() + timeout
     while True:
         with path.open('rb') as stream:
             stream.seek(offset)
@@ -123,32 +126,83 @@ def check_mock_requests(path, offset, expected):
         time.sleep(0.05)
 
 
-def rejected_group(case, group):
+def signed_headers(body, secret, timestamp=None):
+    timestamp = str(int(time.time()) if timestamp is None else timestamp)
+    signature = hmac.new(secret.encode(), timestamp.encode() + b'.' + body, hashlib.sha256).hexdigest()
+    return {'Content-Type': 'application/json', 'X-LB-Timestamp': timestamp, 'X-LB-Signature': 'sha256=' + signature}
+
+
+def rejected_group(case, group, env=None, *, timeout=10):
     settings = case['http_group']
+    env = os.environ if env is None else env
+    kind = case['reject_group']
+    payload = {
+        'session_id': group,
+        'session_type': 'group',
+        'sender': {'id': 'qa-rejected'},
+        'message': [{'type': 'Plain', 'text': settings['message']}],
+    }
+    if kind == 'missing_session':
+        payload.pop('session_id')
+    if kind == 'invalid_message':
+        payload['message'] = 'not a message chain'
+    body = json.dumps(payload).encode()
+    if kind == 'invalid_json':
+        body = b'{'
+    if kind == 'invalid_object':
+        body = b'[]'
+    timestamp = {
+        'expired_signature': int(time.time()) - 600,
+        'future_signature': int(time.time()) + 600,
+        'bad_timestamp': 'invalid',
+    }.get(kind)
     headers = {'Content-Type': 'application/json'}
-    if case['reject_group'] == 'bad_signature':
+    if kind == 'bad_signature':
         headers.update({'X-LB-Timestamp': str(int(time.time())), 'X-LB-Signature': 'sha256=' + '0' * 64})
+    elif kind != 'missing_signature':
+        headers = signed_headers(body, env[settings['secret_env']], timestamp)
+    if kind == 'tampered_body':
+        body += b' '
+    request = urllib.request.Request(settings['url'], headers=headers, data=body)
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    try:
+        with opener.open(request, timeout=timeout):
+            raise ValueError('Invalid bot ingress was accepted')
+    except urllib.error.HTTPError as exc:
+        payload = json.load(exc)
+        return {'http_status': exc.code, 'code': payload.get('code')}
+
+
+def api_probe(case, *, timeout=10):
+    probe = case['http_probe']
+    headers = {} if probe['auth'] == 'missing' else {'X-API-Key': 'qa-intentionally-invalid'}
+    request = urllib.request.Request(probe['url'], headers=headers)
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    try:
+        response = opener.open(request, timeout=timeout)
+    except urllib.error.HTTPError as exc:
+        response = exc
+    with response:
+        return {'http_status': response.status, 'body': json.load(response)}
+
+
+def reset_group(case, group, env, *, timeout=10):
+    body = json.dumps({'session_id': group, 'session_type': 'group'}).encode()
+    settings = case['http_group']
     request = urllib.request.Request(
-        settings['url'],
-        headers=headers,
-        data=json.dumps(
-            {
-                'session_id': group,
-                'session_type': 'group',
-                'sender': {'id': 'qa-rejected'},
-                'message': [{'type': 'Plain', 'text': settings['message']}],
-            }
-        ).encode(),
+        settings['url'].removesuffix('/sync') + '/reset',
+        data=body,
+        headers=signed_headers(body, env[settings['secret_env']]),
     )
     opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
     try:
-        with opener.open(request, timeout=10):
-            raise ValueError('Unsigned/tampered group message was accepted')
+        with opener.open(request, timeout=timeout) as response:
+            result = json.load(response)
     except urllib.error.HTTPError as exc:
-        payload = json.load(exc)
-        if exc.code != 401 or not isinstance(payload, dict) or payload.get('code') != 40101:
-            raise ValueError('Expected HTTP 401 with bad_signature code 40101') from exc
-        return {'http_status': exc.code, 'code': payload['code']}
+        # A received error response violates reset's success contract; do not retry it.
+        raise ValueError(f'Reset returned HTTP {exc.code}; expected 200') from exc
+    check_fields(result, {'code': 0, 'data.session_id': group, 'data.removed': True})
+    return result
 
 
 def verify_cleanup(suite, deleted):
@@ -177,7 +231,7 @@ def verify_cleanup(suite, deleted):
         raise ValueError('Deleted resources remain visible in CLI list readback')
 
 
-def run_smoke(suite):
+def run_cycle(suite, rows, health, cycle):
     # Imported here because Suite dispatches to this module after profile validation.
     from regression_suite import resolve
 
@@ -186,7 +240,10 @@ def run_smoke(suite):
     if not path.is_file():
         raise ValueError('smoke.ledger_db must be an existing target Core SQLite database')
     mock_path = Path(suite.path(settings['mock_records'])) if settings.get('mock_records') else None
-    rows, errors, health = [], [], []
+    errors = []
+    initial_count = len(rows)
+    complete = True
+    execution = getattr(suite, 'execution', None)
     started = time.monotonic()
 
     def redact(value):
@@ -196,9 +253,9 @@ def run_smoke(suite):
 
     def sample_health(phase):
         for name, url in resolve(suite.profile['observe']['health'], suite.variables).items():
-            row = {'phase': phase, 'endpoint': name}
+            row = {'phase': phase, 'cycle': cycle, 'endpoint': name}
             try:
-                row['result'] = read_endpoint_snapshot(url, timeout_seconds=3)
+                row['result'] = read_endpoint_snapshot(url, timeout_seconds=suite.timeout(3))
             except (RuntimeError, OSError, ValueError) as exc:
                 row['error'] = redact(str(exc))
                 errors.append(f'{phase}/{name}: health check failed')
@@ -211,6 +268,9 @@ def run_smoke(suite):
         connection.row_factory = sqlite3.Row
         connection.execute('PRAGMA query_only=ON')
         for original in suite.profile['cases']:
+            if execution and not execution.admitting():
+                complete = False
+                break
             case = resolve(original, suite.variables)
             is_group = 'http_group' in case
             rejected = 'reject_group' in case
@@ -222,40 +282,76 @@ def run_smoke(suite):
             )
             previous = {group: 'none' for group in groups}
             retired = set()
+            reset_done = False
+            history_turns = {group: 0 for group in groups}
             for turn in range(1, (settings.get('group_turns', 2) if is_group and not rejected else 1) + 1):
-                for group in groups:
+                for group_index, group in enumerate(groups):
+                    if execution and not execution.admitting():
+                        complete = False
+                        break
                     row = {
                         'case': case['name'],
+                        'cycle': cycle,
                         'stage': case.get('stage', 'execution'),
                         'status': 'failed',
                         'requires_runner': runner,
                     }
                     if is_group:
-                        row.update(group_id=group, turn=turn, member_id=f'qa-member-{turn % 2:03d}')
+                        row.update(
+                            group_id=group, turn=turn, group_index=group_index, member_id=f'qa-member-{turn % 2:03d}'
+                        )
+                    if case.get('reset_group') and group_index == 1 and turn >= 2 and not reset_done:
+                        row.update(status='skipped', error='Peer reset did not complete; isolation cannot be checked')
+                        rows.append(row)
+                        continue
                     if group in retired:
                         row.update(status='skipped', error='Earlier turn failed or is unknown; group not retried')
                         rows.append(row)
                         continue
                     before = time.monotonic()
                     after_id = connection.execute('SELECT COALESCE(MAX(id),0) FROM agent_run').fetchone()[0]
+                    row['after_id'] = after_id
+                    if runner and is_group:
+                        row['conversation_id'] = 'group_' + group
                     token = uuid4().hex
                     offset = mock_path.stat().st_size if case.get('mock_requests') and mock_path.exists() else 0
                     try:
                         if rejected:
-                            data = checked(row, 'response', lambda: rejected_group(case, group))
+                            data = checked(
+                                row,
+                                'response',
+                                lambda: rejected_group(case, group, suite.env, timeout=suite.timeout(10)),
+                            )
                             count = connection.execute(
-                                'SELECT COUNT(*) FROM agent_run WHERE workspace_id=? AND conversation_id=? AND id>?',
-                                (suite.variables['workspace_uuid'], 'group_' + group, after_id),
+                                'SELECT COUNT(*) FROM agent_run WHERE workspace_id=? AND id>?',
+                                (suite.variables['workspace_uuid'], after_id),
                             ).fetchone()[0]
                             checked(row, 'no_run', lambda: check_fields({'runs': count}, {'runs': 0}))
                         elif is_group:
+                            if case.get('reset_group') and turn == 2 and group_index == 0:
+                                checked(
+                                    row, 'reset', lambda: reset_group(case, group, suite.env, timeout=suite.timeout(10))
+                                )
+                                reset_done = True
+                                history_turns[group] = 0
+                                previous[group] = 'none'
                             data = checked(
                                 row,
                                 'response',
                                 lambda: group_request(
-                                    case, group, row['member_id'], turn, token, previous[group], suite.env
+                                    case,
+                                    group,
+                                    row['member_id'],
+                                    history_turns[group] + 1,
+                                    token,
+                                    previous[group],
+                                    suite.env,
+                                    timeout=suite.timeout(90),
                                 ),
                             )['data']
+                        elif 'http_probe' in case:
+                            data = api_probe(case, timeout=suite.timeout(10))
+                            checked(row, 'response', lambda: check_response(data, case['expect']))
                         else:
                             data = suite.call(case['args'], case.get('body'))
                             checked(row, 'response', lambda: check_response(data, case['expect']))
@@ -273,7 +369,9 @@ def run_smoke(suite):
                             row['checks']['mock_requests'] = checked(
                                 row,
                                 'model_sequence',
-                                lambda: check_mock_requests(mock_path, offset, case['mock_requests']),
+                                lambda: check_mock_requests(
+                                    mock_path, offset, case['mock_requests'], timeout=suite.timeout(3)
+                                ),
                             )
                         if runner:
                             if not isinstance(data, dict):
@@ -281,23 +379,29 @@ def run_smoke(suite):
                             session = data.get('session_id')
                             if not isinstance(session, str) or not session:
                                 raise UnresolvedRun('Reply has no session ID for AgentRun reconciliation')
+                            row['conversation_id'] = 'group_' + session if is_group else session
                             run = completed_run(
                                 connection,
                                 suite.variables['workspace_uuid'],
                                 'group_' + session if is_group else session,
                                 after_id,
-                                settings.get('timeout_seconds', 30),
+                                suite.timeout(settings.get('timeout_seconds', 30)),
                             )
                             row['run'] = run
                             row['checks'].update(checked(row, 'ledger', lambda: check_run(connection, run, case)))
                         row['status'] = 'passed'
                         previous[group] = token
+                        history_turns[group] += 1
                     except urllib.error.HTTPError as exc:
                         row.update(
                             status='unknown' if exc.code >= 500 else 'failed', error=f'HTTP {exc.code}; not retried'
                         )
-                    except (UnresolvedRun, urllib.error.URLError, TimeoutError) as exc:
+                    except TimeoutError as exc:
+                        row.update(status='unknown', timed_out=True, error=redact(str(exc)))
+                    except (UnresolvedRun, urllib.error.URLError, ConnectionError) as exc:
                         row.update(status='unknown', error=redact(str(exc)))
+                        if isinstance(getattr(exc, 'reason', None), TimeoutError):
+                            row['timed_out'] = True
                     except (RuntimeError, ValueError, OSError, KeyError, IndexError, sqlite3.Error) as exc:
                         message = redact(str(exc))
                         row.update(
@@ -306,6 +410,8 @@ def run_smoke(suite):
                             else 'failed',
                             error=message,
                         )
+                        if 'timed out' in message:
+                            row['timed_out'] = True
                     row['seconds'] = time.monotonic() - before
                     if row['status'] != 'passed':
                         errors.append(f'{case["name"]}: {row["status"]}: {row["error"]}')
@@ -321,6 +427,8 @@ def run_smoke(suite):
     sample_health('after')
     return {
         'mode': 'smoke',
+        'cycle_complete': complete,
+        'cycle_requests': len(rows) - initial_count,
         'phases': [],
         'cases': rows,
         'errors': errors,
@@ -333,3 +441,80 @@ def run_smoke(suite):
         'case_count': len(suite.profile['cases']),
         'requests': len(rows),
     }
+
+
+def run_smoke(suite):
+    rows, health, errors, cycles = [], [], [], []
+    execution = getattr(suite, 'execution', None)
+    while True:
+        cycle = len(cycles) + 1
+        if execution:
+            execution.mark(f'cycle-{cycle}')
+        result = run_cycle(suite, rows, health, cycle)
+        errors.extend(result['errors'])
+        cycles.append(
+            {
+                'cycle': cycle,
+                'requests': result['cycle_requests'],
+                'complete': result['cycle_complete'],
+                'elapsed_seconds': result['elapsed_seconds'],
+                'status': dict(Counter(row['status'] for row in rows if row['cycle'] == cycle)),
+            }
+        )
+        suite.write('cycles.json', cycles)
+        if not execution or not execution.admitting() or any(row['status'] == 'unknown' for row in rows):
+            break
+    result['errors'] = errors
+    result['cycles'] = cycles
+    result['first_pass_complete'] = cycles[0]['complete']
+    result['incomplete'] = [
+        f'{row["case"]}: {row["status"]}: {row["error"]}'
+        for row in rows
+        if row['status'] == 'unknown' and not row.get('timed_out')
+    ]
+    if not result['first_pass_complete']:
+        error = 'Required first pass did not finish before the admission deadline'
+        result['errors'].append(error)
+        result['incomplete'].append(error)
+    result['elapsed_seconds'] = sum(cycle['elapsed_seconds'] for cycle in cycles)
+    result['performance_mode'] = 'observe'
+    result['observations'] = {
+        'successful_requests_per_second': sum(row['status'] == 'passed' for row in rows)
+        / max(result['elapsed_seconds'], 0.000001),
+        'step_p95_seconds': {
+            name: percentile([row['seconds'] for row in rows if row['case'] == name and 'seconds' in row], 0.95)
+            for name in dict.fromkeys(row['case'] for row in rows)
+        },
+    }
+    executed = [row for row in rows if row.get('requires_runner') and 'after_id' in row]
+    result['test_data'] = {
+        'observed_conversations': len({row['conversation_id'] for row in executed if 'conversation_id' in row}),
+        'observed_agent_runs': len({row['run']['run_id'] for row in executed if 'run' in row}),
+        'group_sessions': len({row['group_id'] for row in executed if 'group_id' in row}),
+        'owned_api_resources': len(suite.owned),
+    }
+    reconcile_scenario_runs(suite, result)
+    return result
+
+
+def reconcile_scenario_runs(suite, result, *, key='ledger'):
+    """Recheck the originally observed conversations without expanding run ownership."""
+    executed = [row for row in result.get('cases', []) if row.get('requires_runner') and 'after_id' in row]
+    if not executed:
+        return
+    try:
+        result[key] = reconcile_runs(
+            suite.path(suite.profile['smoke']['ledger_db']),
+            suite.variables['workspace_uuid'],
+            min(row['after_id'] for row in executed),
+            {row['conversation_id'] for row in executed if 'conversation_id' in row},
+            {row['run']['run_id'] for row in executed if 'run' in row},
+        )
+    except (OSError, sqlite3.Error) as exc:
+        error = f'Scenario {key} evidence unavailable: {exc}'
+        result['errors'].append(error)
+        result.setdefault('incomplete', []).append(error)
+        return
+    if not result[key]['matched']:
+        error = f'Scenario {key} found extra, missing or non-completed AgentRuns'
+        result['errors'].append(error)

tools/regression_suite.py

@@ -81,6 +81,8 @@ def validate(profile):
         'verification',
         'resource_limits',
         'coverage_suite',
+        'execution',
+        'performance_mode',
     }
     if extra := profile.keys() - allowed:
         raise ValueError(f'unknown profile fields: {sorted(extra)}')
@@ -94,6 +96,20 @@ def validate(profile):
             raise ValueError(f'{field} must be an array of objects')
     if 'coverage_suite' in profile and not re.fullmatch(r'[a-z][a-z0-9_-]*', str(profile['coverage_suite'])):
         raise ValueError('coverage_suite must be a stable suite identifier')
+    if profile.get('performance_mode', 'gate') not in {'gate', 'observe'}:
+        raise ValueError('performance_mode must be gate or observe')
+    execution = profile.get('execution')
+    if execution is not None:
+        if (
+            'smoke' not in profile
+            or not isinstance(execution, dict)
+            or set(execution) != {'total_seconds', 'drain_seconds', 'cooldown_seconds'}
+        ):
+            raise ValueError('execution requires finite smoke cases and total_seconds/drain_seconds/cooldown_seconds')
+        for key, value in execution.items():
+            number(value, f'execution.{key}', minimum=0.001)
+        if execution['drain_seconds'] + execution['cooldown_seconds'] >= execution['total_seconds']:
+            raise ValueError('execution reserves must leave time for required cases')
     cli = profile.get('cli', {})
     if not isinstance(cli, dict) or cli.keys() - {'config', 'executable'}:
         raise ValueError('cli accepts config and executable')
@@ -174,6 +190,8 @@ def validate(profile):
             'fields',
             'forbid',
             'reject_group',
+            'http_probe',
+            'reset_group',
         }:
             raise ValueError('unknown case field')
         if not isinstance(case.get('name'), str) or not case['name'] or case['name'] in names:
@@ -181,7 +199,10 @@ def validate(profile):
         names.add(case['name'])
         if 'stage' in case and (not isinstance(case['stage'], str) or not case['stage']):
             raise ValueError('case stage must be a non-empty string')
-        if any(key in case for key in ('mock_requests', 'reject_group')) and 'smoke' not in profile:
+        if (
+            any(key in case for key in ('mock_requests', 'reject_group', 'http_probe', 'reset_group'))
+            and 'smoke' not in profile
+        ):
             raise ValueError('extended case checks require smoke mode')
         if 'fields' in case and (not isinstance(case['fields'], dict) or not case['fields']):
             raise ValueError('fields must be a non-empty mapping of dotted paths to exact values')
@@ -209,11 +230,37 @@ def validate(profile):
                 number(request['status'], 'mock request status', minimum=100, integer=True)
         if 'reject_group' in case and (
             'http_group' not in case
-            or case['reject_group'] not in ('missing_signature', 'bad_signature')
+            or case['reject_group']
+            not in (
+                'missing_signature',
+                'bad_signature',
+                'expired_signature',
+                'future_signature',
+                'bad_timestamp',
+                'tampered_body',
+                'invalid_json',
+                'invalid_object',
+                'missing_session',
+                'invalid_message',
+            )
             or any(key in case for key in ('box_exec', 'attachment', 'mock_requests'))
         ):
             raise ValueError('reject_group must name a signature rejection on an HTTP group case')
-        if 'http_group' in case:
+        if 'reset_group' in case and (
+            case['reset_group'] is not True or 'http_group' not in case or 'reject_group' in case
+        ):
+            raise ValueError('reset_group requires an accepted group case')
+        if 'http_probe' in case:
+            probe = case['http_probe']
+            if (
+                not isinstance(probe, dict)
+                or set(probe) != {'url', 'auth'}
+                or probe['auth'] not in ('missing', 'invalid')
+                or not resolve(probe['url'], variables).startswith(('http://', 'https://'))
+                or any(key in case for key in ('args', 'body', 'http_group', 'box_exec', 'attachment'))
+            ):
+                raise ValueError('http_probe requires url/auth without CLI or group inputs')
+        elif 'http_group' in case:
             settings = case['http_group']
             if (
                 'args' in case
@@ -347,7 +394,7 @@ def validate(profile):
             raise ValueError(f'unknown case in max_p95_seconds: {name}')
         number(value, 'max_p95_seconds', minimum=0.001)
     if 'resource_limits' in profile:
-        if 'smoke' in profile:
+        if 'smoke' in profile and execution is None:
             raise ValueError('resource_limits requires measured load and cooldown phases')
         from regression_analysis import validate_resource_limits
 
@@ -383,6 +430,14 @@ def validate(profile):
         configured = set(observed.get('pids', {})) | set(observed.get('pid_files', {}))
         if set(profile['resource_limits']['targets']) - configured:
             raise ValueError('Resource limit targets must have configured PIDs or PID files')
+        if execution:
+            limits = profile['resource_limits']
+            if (
+                execution['cooldown_seconds'] < limits['window_seconds']
+                or execution['total_seconds'] - execution['drain_seconds'] - execution['cooldown_seconds']
+                < limits['warmup_seconds'] + limits['window_seconds']
+            ):
+                raise ValueError('execution must cover resource warmup, stable window and cooldown')
         for phase in phases:
             limits = profile['resource_limits']
             if (
@@ -414,6 +469,7 @@ class Suite:
         ]
         self.owned = []
         self.connection = {}
+        self.execution = None
         self.env = dict(os.environ if env is None else env)
         if any('http_group' in case for case in profile['cases']):
             self.variables['group_secret'] = secrets.token_urlsafe(32)
@@ -422,6 +478,9 @@ class Suite:
             value for key, value in self.env.items() if ('API_KEY' in key or key == 'LANGBOT_QA_GROUP_SECRET') and value
         ]
 
+    def timeout(self, maximum):
+        return self.execution.timeout(maximum) if self.execution else maximum
+
     def path(self, value):
         path = Path(resolve(value, self.variables)).expanduser()
         return str(path if path.is_absolute() else Path(self.variables['profile_dir']) / path)
@@ -437,12 +496,26 @@ class Suite:
         return str(path)
 
     def finish(self, result):
+        policy = self.profile.get('cleanup', 'on-success')
+        if policy == 'always' or (policy == 'on-success' and not result['errors']):
+            if (self.output / 'owned.json').exists():
+                try:
+                    result['errors'].extend(self.cleanup())
+                except (RuntimeError, ValueError, OSError, KeyError) as exc:
+                    result['errors'].append(f'cleanup failed: {exc}')
+        if self.execution:
+            self.execution.finish(result)
         if self.profile.get('coverage_suite'):
             from regression_coverage import write_report
 
             coverage = write_report([self.output], self.output)
             result['errors'].extend(coverage['evidence_errors'])
+            result.setdefault('incomplete', []).extend(coverage['evidence_errors'])
         result['passed'] = not result['errors']
+        if result['errors']:
+            result['outcome'] = 'FAIL' if set(result['errors']) - set(result.get('incomplete', [])) else 'INCOMPLETE'
+        else:
+            result['outcome'] = 'PASS'
         self.write('summary.json', result)
 
     def call(self, args, body=None, *, defer_bot_readback=False):
@@ -450,7 +523,7 @@ class Suite:
         if body is not None:
             command += ['--file', self.write('request.json', body)]
         try:
-            result = subprocess.run(command, capture_output=True, text=True, timeout=360, env=self.env)
+            result = subprocess.run(command, capture_output=True, text=True, timeout=self.timeout(360), env=self.env)
         except subprocess.TimeoutExpired as exc:
             raise RuntimeError('lbctl timed out; outcome unknown, no automatic retry') from exc
         if result.returncode:
@@ -486,7 +559,7 @@ class Suite:
         )
         opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
         try:
-            with opener.open(request, timeout=30) as response:
+            with opener.open(request, timeout=self.timeout(30)) as response:
                 payload = json.load(response)
         except (urllib.error.URLError, OSError) as exc:
             raise RuntimeError('Bot readback request failed; the creation was not retried') from exc
@@ -523,6 +596,11 @@ class Suite:
             self.write('environment.json', {'identity': identity, 'capabilities': capabilities, 'plugins': plugins})
             if missing:
                 raise RuntimeError(f'Required plugins are not initialized: {sorted(missing)}')
+        if self.profile.get('execution'):
+            from regression_scenarios import ExecutionWindow
+
+            self.execution = ExecutionWindow(self)
+            self.execution.start()
         for item in self.profile.get('resources', []):
             kind = item['kind']
             body = resolve(item['body'], self.variables)
@@ -588,6 +666,8 @@ class Suite:
                 self.write('warmup.json', warmups)
 
     def cleanup(self):
+        if self.execution:
+            self.execution.mark('cleanup')
         saved = json.loads((self.output / 'owned.json').read_text())
         identity = self.call(['whoami'])
         if saved['workspace_uuid'] != identity['workspace_uuid']:
@@ -715,6 +795,9 @@ class Suite:
                 errors.extend(f'{phase["name"]}: {error}' for error in summary['resources']['errors'])
             if result.returncode:
                 errors.append(f'{phase["name"]}: business or health/resource checks failed')
+            summary['performance_mode'] = self.profile.get('performance_mode', 'gate')
+            if summary['performance_mode'] == 'observe':
+                continue
             thresholds = self.profile.get('thresholds', {})
             if summary['achieved_rps'] < phase['rps'] * thresholds.get('min_rps_ratio', 0):
                 errors.append(f'{phase["name"]}: achieved RPS below configured threshold')
@@ -783,13 +866,6 @@ def main():
         finally:
             if not completed and not result['errors']:
                 result['errors'].append('Run aborted before completion; inspect logs and retained resources.')
-            policy = profile.get('cleanup', 'on-success')
-            if policy == 'always' or (policy == 'on-success' and not result['errors']):
-                if (args.output / 'owned.json').exists():
-                    try:
-                        result['errors'].extend(suite.cleanup())
-                    except (RuntimeError, ValueError, OSError, KeyError) as exc:
-                        result['errors'].append(f'cleanup failed: {exc}')
             suite.finish(result)
         print(json.dumps(result, ensure_ascii=False, indent=2))
         if not result['passed']:

tools/regression_traffic.py

@@ -87,7 +87,7 @@ def sample_containers(configured, data_root, run=subprocess.run, discover=owned_
     return observation
 
 
-def group_request(case, group_id, member_id, turn, token, previous, env=None):
+def group_request(case, group_id, member_id, turn, token, previous, env=None, *, timeout=90):
     """Send one signed GroupMessage through the existing HTTP Bot adapter."""
     settings = case['http_group']
     secret = (os.environ if env is None else env)[settings['secret_env']]
@@ -113,7 +113,7 @@ def group_request(case, group_id, member_id, turn, token, previous, env=None):
         },
     )
     opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
-    with opener.open(request, timeout=90) as response:
+    with opener.open(request, timeout=timeout) as response:
         payload = json.load(response)
         if response.status != 200 or not isinstance(payload, dict) or payload.get('code') != 0:
             raise ValueError('group ingress did not return a completed response')

tools/test_regression_coverage.py

@@ -100,6 +100,35 @@ class CoverageTest(unittest.TestCase):
         self.assertEqual(data['contracts'][0]['status'], 'not_run')
         self.assertEqual(data['unclassified_cases'], [{'profile': 'smoke', 'case': 'new_case'}])
 
+    def test_reset_failure_and_peer_skip_cannot_be_hidden_by_first_turns(self):
+        c = load_catalog()
+        rows = [
+            {
+                'profile': 'scenarios',
+                'case': 'group_reset_isolation',
+                'status': 'passed',
+                'group_index': group,
+                'turn': 1,
+                'assertions': {'response': 'passed'},
+            }
+            for group in (0, 1)
+        ]
+        rows += [
+            {
+                'profile': 'scenarios',
+                'case': 'group_reset_isolation',
+                'status': 'failed',
+                'group_index': 0,
+                'turn': 2,
+                'assertions': {'reset': 'failed'},
+            },
+            {'profile': 'scenarios', 'case': 'group_reset_isolation', 'status': 'skipped', 'group_index': 1, 'turn': 2},
+        ]
+        selected = [
+            x for x in report(c, rows)['contracts'] if any(b['case'] == 'group_reset_isolation' for b in x['bindings'])
+        ]
+        self.assertEqual(sorted(x['status'] for x in selected), ['failed', 'skipped'])
+
     def test_partial_evidence_produces_an_error_and_a_report(self):
         with tempfile.TemporaryDirectory() as d:
             root = Path(d)
@@ -260,7 +289,13 @@ class Routes:
                 name = binding['profile']
                 if name not in profiles:
                     path = Path(__file__).with_name('regression_profiles') / f'{name}.json'
-                    profiles[name] = {x['name']: x for x in json.loads(path.read_text())['cases']}
+                    if name == 'scenarios':
+                        from regression_scenarios import expanded_profile
+
+                        profile = expanded_profile()
+                    else:
+                        profile = json.loads(path.read_text())
+                    profiles[name] = {x['name']: x for x in profile['cases']}
                 self.assertIn(binding['case'], profiles[name])
                 self.assertTrue(binding['checks'])
         broken = copy.deepcopy(c)

tools/test_regression_job.py

@@ -14,6 +14,8 @@ from unittest.mock import patch
 import yaml
 
 from regression_job import Job, load_profile, main
+from regression_scenarios import ExecutionWindow
+from regression_suite import Suite
 from test_regression_suite import FAKE_CLI, profile
 
 
@@ -177,6 +179,74 @@ class JobTests(unittest.TestCase):
             server.server_close()
             thread.join()
 
+    def test_shared_finish_cleans_owned_resources_before_environment_teardown(self):
+        for failure in ('prepare', 'finish-error', 'finish-raises'):
+            with self.subTest(failure=failure), tempfile.TemporaryDirectory() as folder:
+                root = Path(folder)
+                current = profile()
+                current.pop('phases')
+                current.update(
+                    smoke={'ledger_db': 'unused.db'},
+                    cleanup='always',
+                    execution={'total_seconds': 1, 'drain_seconds': 0.1, 'cooldown_seconds': 0.01},
+                )
+                job = Job({'cleanup': 'always'}, current, root / 'profile.json', root / 'job')
+                events, visible = [], ['owned']
+
+                def prepare():
+                    destination = job.output / 'suite'
+                    destination.mkdir()
+                    suite = job.suite = Suite(current, root / 'profile.json', destination)
+                    suite.variables['workspace_uuid'] = 'workspace'
+                    suite.owned = [{'kind': 'provider', 'uuid': 'owned'}]
+                    suite.journal()
+
+                    def call(args, body=None):
+                        if args == ['whoami']:
+                            return {'workspace_uuid': 'workspace'}
+                        if args == ['provider', 'delete', 'owned', '--yes']:
+                            events.append('delete-owned')
+                            visible.remove('owned')
+                            return {}
+                        if args == ['provider', 'list']:
+                            return {'providers': [{'uuid': item} for item in visible]}
+                        raise AssertionError(args)
+
+                    suite.call = call
+                    suite.run = lambda: {'errors': [], 'phases': []}
+                    suite.execution = ExecutionWindow(suite)
+                    suite.execution.start()
+                    if failure == 'prepare':
+                        raise RuntimeError('prepare failed after creating resource')
+                    finish = suite.finish
+
+                    def failing_finish(summary):
+                        if failure == 'finish-error':
+                            suite.execution.errors.append('health failed during execution')
+                        finish(summary)
+                        if failure == 'finish-raises':
+                            raise TypeError('finalization defect')
+
+                    suite.finish = failing_finish
+                    return suite
+
+                with (
+                    patch.object(job, 'prepare', side_effect=prepare),
+                    patch.object(job, 'collect', side_effect=lambda: events.append('collect')),
+                    patch.object(job, 'cleanup', side_effect=lambda: events.append('teardown')),
+                    patch('regression_traffic.read_endpoint_snapshot', return_value={'http.ok': 1}),
+                ):
+                    result = job.run()
+                self.assertFalse(result['passed'])
+                self.assertFalse(result['suite']['passed'])
+                self.assertEqual(result['suite']['outcome'], 'FAIL')
+                self.assertFalse(job.suite.execution.thread.is_alive())
+                self.assertEqual(events, ['delete-owned', 'collect', 'teardown'])
+                self.assertEqual(json.loads((job.output / 'suite/owned.json').read_text())['resources'], [])
+                self.assertIn('cleanup', [phase['stage'] for phase in result['suite']['execution']['phases']])
+                self.assertEqual(result['errors'], result['suite']['errors'])
+                self.assertEqual(result['completed'], failure != 'prepare')
+
     def test_cleanup_refuses_changed_deployment(self):
         with tempfile.TemporaryDirectory() as folder:
             root = Path(folder)

tools/test_regression_scenarios.py

@@ -0,0 +1,234 @@
+"""Exercise the real finite runner with a controlled clock and an independent SQLite ledger."""
+
+import json
+from pathlib import Path
+import sqlite3
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from regression_scenarios import ExecutionWindow, expanded_profile
+from regression_smoke import run_smoke
+from regression_suite import Suite, validate
+
+
+class ScenarioExecutionTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name)
+        with sqlite3.connect(self.root / 'ledger.db') as connection:
+            connection.executescript("""
+                CREATE TABLE agent_run (id INTEGER, run_id TEXT, workspace_id TEXT,
+                    conversation_id TEXT, status TEXT, status_reason TEXT, finished_at TEXT);
+                CREATE TABLE agent_run_event (run_id TEXT, sequence INTEGER, type TEXT, data_json TEXT);
+            """)
+        self.profile = {
+            'version': 1,
+            'cli': {'config': 'context.yaml'},
+            'smoke': {'ledger_db': 'ledger.db'},
+            'cases': [{'name': name, 'args': ['provider', 'list'], 'expect': 'OK'} for name in ('a', 'b', 'c')],
+            'observe': {'health': {'core': 'http://unused.invalid/healthz'}},
+            'execution': {'total_seconds': 100, 'drain_seconds': 10, 'cooldown_seconds': 10},
+        }
+        self.suite = Suite(self.profile, self.root / 'profile.json', self.root)
+        self.suite.variables['workspace_uuid'] = 'test'
+        self.now = 0
+        self.clock = patch('time.monotonic', side_effect=lambda: self.now)
+        self.clock.start()
+        self.addCleanup(self.clock.stop)
+        self.suite.execution = ExecutionWindow(self.suite)
+
+    def execute(self, outputs):
+        calls = []
+
+        def call(args, body=None):
+            calls.append(list(args))
+            self.now += 20
+            value = next(outputs)
+            if isinstance(value, Exception):
+                raise value
+            return value
+
+        self.suite.call = call
+        with patch('regression_smoke.read_endpoint_snapshot', return_value={'http.ok': 1}):
+            result = run_smoke(self.suite)
+        return result, calls
+
+    def test_required_first_pass_then_partial_optional_cycle_preserves_evidence(self):
+        result, calls = self.execute(iter(['OK'] * 4))
+        self.assertEqual(len(calls), 4)
+        self.assertTrue(result['first_pass_complete'])
+        self.assertEqual(result['errors'], [])
+        self.assertEqual(
+            [(row['case'], row['cycle']) for row in result['cases']], [('a', 1), ('b', 1), ('c', 1), ('a', 2)]
+        )
+        self.assertEqual([cycle['complete'] for cycle in result['cycles']], [True, False])
+        self.assertEqual(json.loads((self.root / 'cases.json').read_text()), result['cases'])
+
+    def test_first_pass_shortfall_is_incomplete(self):
+        self.profile['execution']['total_seconds'] = 50
+        self.suite.execution = ExecutionWindow(self.suite)
+        result, calls = self.execute(iter(['OK'] * 2))
+        self.assertEqual(len(calls), 2)
+        self.assertFalse(result['first_pass_complete'])
+        self.assertIn('Required first pass', result['incomplete'][0])
+        self.assertTrue(result['errors'])
+
+    def test_later_success_does_not_erase_failure_and_unknown_stops_repetition(self):
+        result, _ = self.execute(iter(['wrong', 'OK', 'OK', 'OK']))
+        self.assertEqual(result['status'], {'failed': 1, 'passed': 3})
+        self.assertEqual(result['cases'][0]['assertions']['response'], 'failed')
+        self.assertTrue(result['errors'])
+        self.now = 0
+        result, calls = self.execute(iter([RuntimeError('outcome unknown'), 'OK', 'OK']))
+        self.assertEqual(len(calls), 3)
+        self.assertEqual(result['status'], {'unknown': 1, 'passed': 2})
+        self.assertTrue(result['incomplete'])
+        self.assertEqual(len(result['cycles']), 1)
+
+    def test_connection_reset_is_unknown_and_does_not_start_another_cycle(self):
+        self.profile['cases'][0] = {
+            'name': 'transport',
+            'http_probe': {'url': 'http://unused.invalid', 'auth': 'missing'},
+            'expect': '401',
+        }
+        with patch('regression_smoke.api_probe', side_effect=ConnectionResetError('peer reset')) as probe:
+            result, calls = self.execute(iter(['OK'] * 4))
+        self.assertEqual(probe.call_count, 1)
+        self.assertEqual(len(calls), 2)
+        self.assertEqual(result['status'], {'unknown': 1, 'passed': 2})
+        self.assertEqual(len(result['cycles']), 1)
+        self.assertTrue(result['first_pass_complete'])
+        self.assertEqual(result['errors'], result['incomplete'])
+
+    def test_cooldown_reconciles_late_runs_against_original_scope(self):
+        self.profile['cases'] = [{'name': 'chat', 'args': ['pipeline', 'run', 'owned'], 'expect': 'OK'}]
+        self.profile['execution']['total_seconds'] = 50
+        path = self.root / 'ledger.db'
+
+        def call(args, body=None):
+            self.now = 30
+            with sqlite3.connect(path) as connection:
+                connection.execute("INSERT INTO agent_run VALUES (1,'expected','test','session','completed','','now')")
+                connection.execute("INSERT INTO agent_run_event VALUES ('expected',1,'run.completed','{}')")
+            return {'session_id': 'session', 'reply': 'OK'}
+
+        self.suite.call = call
+        for late_status in (None, 'completed', 'running'):
+            with self.subTest(late_status=late_status):
+                self.now = 0
+                with sqlite3.connect(path) as connection:
+                    connection.execute('DELETE FROM agent_run')
+                    connection.execute('DELETE FROM agent_run_event')
+                window = self.suite.execution = ExecutionWindow(self.suite)
+                with patch('regression_smoke.read_endpoint_snapshot', return_value={'http.ok': 1}):
+                    result = run_smoke(self.suite)
+                self.assertTrue(result['ledger']['matched'])
+                self.assertEqual(result['errors'], [])
+
+                def cooldown(_seconds):
+                    self.assertEqual(window.stage, 'cooldown')
+                    with sqlite3.connect(path) as connection:
+                        connection.execute(
+                            "INSERT INTO agent_run VALUES (2,'other-ws','other','session','running','',NULL)"
+                        )
+                        connection.execute(
+                            "INSERT INTO agent_run VALUES (3,'other-session','test','other','running','',NULL)"
+                        )
+                        if late_status:
+                            connection.execute(
+                                "INSERT INTO agent_run VALUES (4,'late','test','session',?,'',?)",
+                                (late_status, 'now' if late_status == 'completed' else None),
+                            )
+                    self.now = window.deadline - 1
+
+                with (
+                    patch.object(window.stop, 'wait', side_effect=cooldown),
+                    patch.object(window.thread, 'join'),
+                    patch.object(window.thread, 'is_alive', return_value=False),
+                ):
+                    window.finish(result)
+                self.assertTrue(result['ledger']['matched'])
+                self.assertEqual(result['ledger_closeout']['matched'], late_status is None)
+                self.assertEqual(result['ledger_closeout']['extra'], ['late'] if late_status else [])
+                self.assertEqual(
+                    result['ledger_closeout']['not_completed'], ['late'] if late_status == 'running' else []
+                )
+                self.assertEqual(bool(result['errors']), late_status is not None)
+
+    def test_early_abort_waits_only_for_configured_cooldown(self):
+        window = self.suite.execution
+        self.now = 5
+        result = {'errors': ['Preparation failed']}
+
+        def cooldown(seconds):
+            self.assertEqual(seconds, 10)
+            self.now += seconds
+
+        with (
+            patch.object(window.stop, 'wait', side_effect=cooldown),
+            patch.object(window.thread, 'join'),
+            patch.object(window.thread, 'is_alive', return_value=False),
+        ):
+            window.finish(result)
+        self.assertEqual(result['errors'], ['Preparation failed'])
+        self.assertEqual(result['execution']['elapsed_seconds'], 15)
+        self.assertLess(self.now, window.deadline)
+
+    def test_request_timeout_is_unknown_but_remains_a_failure(self):
+        result, _ = self.execute(iter([TimeoutError('request timed out'), 'OK', 'OK']))
+        self.assertEqual(result['cases'][0]['status'], 'unknown')
+        self.assertTrue(result['cases'][0]['timed_out'])
+        self.assertTrue(set(result['errors']) - set(result['incomplete']))
+
+    def test_cli_timeout_shrinks_to_shared_deadline_and_cleanup_reuses_it(self):
+        self.now = 87
+        value = subprocess.CompletedProcess([], 0, '{"data": {}}', '')
+        with patch('regression_suite.subprocess.run', return_value=value) as run:
+            self.suite.call(['provider', 'list'])
+            self.assertEqual(run.call_args.kwargs['timeout'], 3)
+            self.now = 90
+            with self.assertRaises(TimeoutError):
+                self.suite.call(['provider', 'list'])
+            self.assertEqual(run.call_count, 1)
+        (self.root / 'owned.json').write_text(json.dumps({'workspace_uuid': 'test', 'resources': []}))
+        with self.assertRaises(TimeoutError):
+            self.suite.cleanup()
+        self.assertEqual(self.suite.execution.stage, 'cleanup')
+
+    def test_unavailable_process_fails_observation_mode_and_retains_sample(self):
+        self.clock.stop()
+        self.profile['performance_mode'] = 'observe'
+        self.profile['execution'] = {'total_seconds': 0.15, 'drain_seconds': 0.02, 'cooldown_seconds': 0.02}
+        self.profile['observe']['pids'] = {'missing': 999999999}
+        window = ExecutionWindow(self.suite)
+        result = {'errors': []}
+        with patch('regression_traffic.read_endpoint_snapshot', return_value={'http.ok': 1}):
+            window.start()
+            window.finish(result)
+        self.assertTrue(any('process sample failed' in error for error in result['errors']))
+        self.assertFalse(window.thread.is_alive())
+        rows = [json.loads(line) for line in (self.root / 'resources.jsonl').read_text().splitlines()]
+        self.assertIn('error', rows[0]['targets']['missing'])
+        self.assertEqual(
+            [phase['stage'] for phase in result['execution']['phases']], ['prepare', 'cooldown', 'finished']
+        )
+
+    def test_expanded_profile_and_budget_validation(self):
+        expanded = expanded_profile()
+        self.assertEqual(len(expanded['cases']), 46)
+        validate(expanded)
+        validate(self.profile)
+        for execution in (
+            {},
+            {**self.profile['execution'], 'total_seconds': 20},
+            {**self.profile['execution'], 'drain_seconds': float('nan')},
+        ):
+            with self.subTest(execution=execution), self.assertRaises(ValueError):
+                validate({**self.profile, 'execution': execution})
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/test_regression_smoke.py

@@ -8,6 +8,8 @@ import sqlite3
 import tempfile
 from types import SimpleNamespace
 import unittest
+from unittest.mock import patch
+from urllib.error import HTTPError
 
 from regression_smoke import (
     UnresolvedRun,
@@ -17,6 +19,7 @@ from regression_smoke import (
     check_run,
     completed_run,
     verify_cleanup,
+    reset_group,
 )
 from regression_suite import validate
 
@@ -37,6 +40,16 @@ class SmokeOracleTest(unittest.TestCase):
     def run_row(self, workspace='workspace', after=0):
         return completed_run(self.db, workspace, 'conversation', after, 0)
 
+    def test_received_reset_error_is_a_failed_contract_not_unknown_transport(self):
+        case = {'http_group': {'url': 'http://localhost/bots/test/sync', 'secret_env': 'SECRET'}}
+        with patch('regression_smoke.urllib.request.build_opener') as build:
+            build.return_value.open.side_effect = HTTPError('reset', 500, 'error', {}, None)
+            with self.assertRaisesRegex(ValueError, 'Reset returned HTTP 500'):
+                reset_group(case, 'group', {'SECRET': 'test'})
+            build.return_value.open.side_effect = TimeoutError('transport')
+            with self.assertRaises(TimeoutError):
+                reset_group(case, 'group', {'SECRET': 'test'})
+
     def test_terminal_status_and_event_are_both_required(self):
         self.assertTrue(check_run(self.db, self.run_row(), {})['terminal_event'])
         self.db.execute("UPDATE agent_run SET status='failed'")

tools/test_regression_suite.py

@@ -282,16 +282,24 @@ class ProfileTests(unittest.TestCase):
                     str(config),
                 ]
                 environment = {**os.environ, 'LBCTL_TEST_ROOT': folder}
-                for enforce_latency in (False, True):
+                for enforce_latency, mode, wrong_reply in (
+                    (False, 'gate', False),
+                    (True, 'gate', False),
+                    (True, 'observe', False),
+                    (True, 'observe', True),
+                ):
+                    current['performance_mode'] = mode
+                    current['cases'][0]['expect'] = 'WRONG_RESULT' if wrong_reply else 'TEST_OK'
+                    expected_failure = wrong_reply or (enforce_latency and mode == 'gate')
                     current['thresholds'] = {'max_p95_seconds': {'chat': 0.001}} if enforce_latency else {}
                     config.write_text(json.dumps(current))
-                    output = root / ('failure' if enforce_latency else 'success')
+                    output = root / f'{enforce_latency}-{mode}-{wrong_reply}'
                     result = subprocess.run(
                         [*command, '--output', str(output)], env=environment, capture_output=True, text=True, timeout=30
                     )
                     summary = json.loads((output / 'summary.json').read_text())
-                    self.assertEqual(result.returncode, int(enforce_latency), result.stdout + result.stderr)
-                    self.assertEqual(summary['passed'], not enforce_latency)
+                    self.assertEqual(result.returncode, int(expected_failure), result.stdout + result.stderr)
+                    self.assertEqual(summary['passed'], not expected_failure)
                     self.assertEqual(len(summary['phases']), 2)
                     for phase in summary['phases']:
                         self.assertGreater(phase['achieved_rps'], 0)