From 27f7e96b362c6f69c93d6c718998034164a7309f Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <60681390+huanghuoguoguo@users.noreply.github.com> Date: Wed, 28 Oct 2026 12:02:56 +0800 Subject: [PATCH] test(regression): expand finite scenarios and share bounded execution lifecycle --- tools/README.md | 69 ++++++++ tools/regression_analysis.py | 11 +- tools/regression_contracts.json | 155 ++++++++++++++---- tools/regression_coverage.py | 4 + tools/regression_job.py | 23 ++- tools/regression_scenarios.py | 240 +++++++++++++++++++++++++++ tools/regression_smoke.py | 253 +++++++++++++++++++++++++---- tools/regression_suite.py | 102 ++++++++++-- tools/regression_traffic.py | 4 +- tools/test_regression_coverage.py | 37 ++++- tools/test_regression_job.py | 70 ++++++++ tools/test_regression_scenarios.py | 234 ++++++++++++++++++++++++++ tools/test_regression_smoke.py | 13 ++ tools/test_regression_suite.py | 16 +- 14 files changed, 1132 insertions(+), 99 deletions(-) create mode 100644 tools/regression_scenarios.py create mode 100644 tools/test_regression_scenarios.py diff --git a/tools/README.md b/tools/README.md index 3b33cfcc7..1ec7156ad 100644 --- a/tools/README.md +++ b/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. diff --git a/tools/regression_analysis.py b/tools/regression_analysis.py index f643fa760..42222cdf1 100644 --- a/tools/regression_analysis.py +++ b/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 diff --git a/tools/regression_contracts.json b/tools/regression_contracts.json index 22334eec1..048355c99 100644 --- a/tools/regression_contracts.json +++ b/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/"] }, {"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/"] }, {"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/"] }, {"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/"] }, {"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//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//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//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//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//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//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//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//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//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//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//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//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//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//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//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//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//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"]}] } ] } diff --git a/tools/regression_coverage.py b/tools/regression_coverage.py index 73ed9232d..4efc732cf 100644 --- a/tools/regression_coverage.py +++ b/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']] diff --git a/tools/regression_job.py b/tools/regression_job.py index 22b5b7c85..638b822aa 100644 --- a/tools/regression_job.py +++ b/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: diff --git a/tools/regression_scenarios.py b/tools/regression_scenarios.py new file mode 100644 index 000000000..f95fa7598 --- /dev/null +++ b/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, + } diff --git a/tools/regression_smoke.py b/tools/regression_smoke.py index 7734ef87c..c996c0ff5 100644 --- a/tools/regression_smoke.py +++ b/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) diff --git a/tools/regression_suite.py b/tools/regression_suite.py index fbab54989..73c152689 100644 --- a/tools/regression_suite.py +++ b/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']: diff --git a/tools/regression_traffic.py b/tools/regression_traffic.py index fdf6d8518..aa1409951 100644 --- a/tools/regression_traffic.py +++ b/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') diff --git a/tools/test_regression_coverage.py b/tools/test_regression_coverage.py index 99d6f324b..163f8c68b 100644 --- a/tools/test_regression_coverage.py +++ b/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) diff --git a/tools/test_regression_job.py b/tools/test_regression_job.py index 40f75ad32..3cfdd0d7e 100644 --- a/tools/test_regression_job.py +++ b/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) diff --git a/tools/test_regression_scenarios.py b/tools/test_regression_scenarios.py new file mode 100644 index 000000000..9d74d4d45 --- /dev/null +++ b/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() diff --git a/tools/test_regression_smoke.py b/tools/test_regression_smoke.py index b87ae26ae..faf7ed79a 100644 --- a/tools/test_regression_smoke.py +++ b/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'") diff --git a/tools/test_regression_suite.py b/tools/test_regression_suite.py index ae295588a..6d8b59a26 100644 --- a/tools/test_regression_suite.py +++ b/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)