233 lines
13 KiB
Python
233 lines
13 KiB
Python
"""A closed batch deadline keeps only a proved prefix, without a success/retry."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from copy import deepcopy
|
|
|
|
import pytest
|
|
|
|
from apps.server.tests.test_discovery import item, search_candidate
|
|
from apps.server.tests.test_learning import client, initialize
|
|
from apps.server.tests.test_postgres import pg_client
|
|
from apps.server.tests.test_reading import COMMENTS, NOTE_ID, NOTE_URL, TITLE, comment_output, detail_output, learned_detail
|
|
from apps.server.worker import BrowserWorker, WorkerError
|
|
|
|
|
|
@pytest.fixture(params=["client", "pg_client"])
|
|
def batch_budget_client(request):
|
|
return request.getfixturevalue(request.param)
|
|
|
|
|
|
def test_batch_deadline_preserves_proved_prefix_revokes_ready_and_never_retries(batch_budget_client):
|
|
client = batch_budget_client
|
|
initialize(client)
|
|
_, search = search_candidate(client, [item(TITLE)])
|
|
response = client.post("/api/discovery/runs", json={"search_candidate_id": search["id"], "keyword": "准备装修"})
|
|
assert response.status_code == 201 and response.json()["status"] == "completed", response.text
|
|
run = response.json()
|
|
task, candidate, _ = learned_detail(client, seek=True)
|
|
task_endpoint = "/api/learning/tasks/" + task["id"]
|
|
endpoint = "/api/discovery/runs/" + run["id"] + "/readings"
|
|
payload = {"detail_candidate_id": candidate["id"], "note_id": NOTE_ID}
|
|
response = client.post(endpoint, json=payload)
|
|
assert response.status_code == 201 and response.json()["status"] == "completed", response.text
|
|
original_reading = response.json()
|
|
store = client.app_reference.state.store
|
|
originals = {
|
|
"candidate": store.one("SELECT digest,manifest_json FROM candidates WHERE id=$1", (candidate["id"],)),
|
|
"run": store.one("SELECT data_json FROM discovery_runs WHERE id=$1", (run["id"],)),
|
|
"reading": store.one("SELECT data_json FROM reading_runs WHERE id=$1", (original_reading["id"],)),
|
|
"validation": store.one("SELECT data_json FROM validations WHERE id=$1", (original_reading["validation_id"],)),
|
|
}
|
|
worker = client.worker_reference
|
|
original_call = worker.call
|
|
completed_prefix = comment_output()
|
|
completed_prefix["batch_summary"].update(duplicates_removed=1, stop_reason="time_budget")
|
|
completed_prefix["skipped_incomplete_count"] = 1
|
|
expected_timings = []
|
|
|
|
async def deadline_after_completed_batch(method, params):
|
|
value = await original_call(method, params)
|
|
if method == "verify" and params.get("note_binding"):
|
|
# This is a contract substitute: one complete, captured, source-bound
|
|
# batch preceded a later interruption. It performs no browser work.
|
|
batch_capture = worker.capture(params["task_id"])
|
|
batch_capture.update(url=NOTE_URL, text="本机合成的完整首层评论批次")
|
|
value["evidence"].append(batch_capture)
|
|
expected_timings[:] = [
|
|
{"step_position": index, "kind": step["kind"], "status": "passed", "duration_ms": 10,
|
|
"phase_ms": {"execute": 4, "capture_after": 6}}
|
|
for index, step in enumerate(params["steps"][:-1], 1)
|
|
] + [{"step_position": len(params["steps"]), "kind": "extract", "status": "interrupted",
|
|
"duration_ms": 15000, "phase_ms": {"execute": 80, "batch_capture": 14920}}]
|
|
value.update(
|
|
passed=False,
|
|
outputs={"note_detail": detail_output(), "comments": deepcopy(completed_prefix)},
|
|
errors=["批次预算已到，仅保留此前完整取证的评论前缀"],
|
|
error_codes=["BATCH_BUDGET"],
|
|
session_closed=True,
|
|
step_timings=deepcopy(expected_timings),
|
|
verify_timing={"duration_ms": 15200, "entry_ms": 100, "close_ms": 30},
|
|
)
|
|
worker.sessions.pop(params["task_id"], None) # Actual boolean close acknowledgment.
|
|
return value
|
|
|
|
worker.call = deadline_after_completed_batch
|
|
response = client.post(endpoint, json=payload)
|
|
assert response.status_code == 201, response.text
|
|
partial = response.json()
|
|
assert partial["status"] == "partial" and partial["model_connected"] is False
|
|
assert partial["detail"] == detail_output() and partial["comments"] == completed_prefix
|
|
assert partial["comments"]["count"] == 1 and partial["comments"]["truncated"] is True
|
|
assert partial["comments"]["batch_summary"]["stop_reason"] == "time_budget"
|
|
assert partial["error_codes"] == ["BATCH_BUDGET"] and partial["errors"]
|
|
assert partial["note_binding_result"] == {
|
|
"note_id": NOTE_ID, "href_checked": True, "url_checked": True, "source_url": NOTE_URL, "title": TITLE,
|
|
}
|
|
assert partial["seek_results"][0]["target_reached"] is True
|
|
assert partial["step_timings"] == expected_timings
|
|
assert partial["verify_timing"] == {"duration_ms": 15200, "entry_ms": 100, "close_ms": 30}
|
|
assert len(partial["evidence"]) == 2
|
|
assert all(client.get(evidence["screenshot_url"]).status_code == 200 for evidence in partial["evidence"])
|
|
|
|
after = client.get(task_endpoint).json()
|
|
assert after["status"] == "paused" and after["browser_status"] == "closed"
|
|
assert after["browser_readiness"]["ready"] is False
|
|
assert after["browser_readiness"]["next_action"] == "open_browser"
|
|
validation = next(value for value in after["validations"] if value["id"] == partial["validation_id"])
|
|
assert validation["status"] == "failed" and validation["error_codes"] == ["BATCH_BUDGET"]
|
|
assert validation["environment"]["browser_session_closed"] is True
|
|
assert validation["environment"]["verify_budget_ms"] == 95000
|
|
assert validation["actual_outputs"] == {"note_detail": detail_output(), "comments": completed_prefix}
|
|
assert validation["evidence"] == partial["evidence"]
|
|
assert validation["step_timings"] == partial["step_timings"]
|
|
assert validation["verify_timing"] == partial["verify_timing"]
|
|
# Fixed output/source assertions may pass on this finite prefix. The failed
|
|
# worker execution must still prevent completed reading or a passed trial.
|
|
checks = {check["name"]: check["passed"] for check in validation["assertion_checks"]}
|
|
assert all(checks[name] for name in (
|
|
"output_schema:note_detail", "output_nonempty:note_detail", "output_schema:comments",
|
|
"output_nonempty:comments", "note_source_binding",
|
|
))
|
|
|
|
calls_before = len(worker.calls)
|
|
response = client.post(endpoint, json=payload)
|
|
assert response.status_code == 201, response.text
|
|
stopped = response.json()
|
|
assert stopped["status"] == "failed" and stopped["validation_id"] is None
|
|
assert stopped["detail"] is None and stopped["comments"] is None
|
|
assert stopped["error_codes"] == ["browser_session_required"]
|
|
assert worker.calls[calls_before:] == [] # No status/open/ready/verify or implicit retry.
|
|
assert client.get("/api/discovery/readings/" + partial["id"]).json() == partial
|
|
assert client.get("/api/discovery/readings/" + original_reading["id"]).json() == original_reading
|
|
assert store.one("SELECT digest,manifest_json FROM candidates WHERE id=$1", (candidate["id"],)) == originals["candidate"]
|
|
assert store.one("SELECT data_json FROM discovery_runs WHERE id=$1", (run["id"],)) == originals["run"]
|
|
assert store.one("SELECT data_json FROM reading_runs WHERE id=$1", (original_reading["id"],)) == originals["reading"]
|
|
assert store.one("SELECT data_json FROM validations WHERE id=$1", (original_reading["validation_id"],)) == originals["validation"]
|
|
|
|
|
|
@pytest.mark.parametrize("close_ack", [True, "true", False], ids=["closed", "string-is-not-ack", "not-closed"])
|
|
def test_direct_batch_failure_only_boolean_close_revokes_ready_without_recording_or_retry(batch_budget_client, close_ack):
|
|
client = batch_budget_client
|
|
initialize(client)
|
|
# The reading helper normally follows search_candidate, whose fake capture
|
|
# declares the ordinary search input. This direct-action fixture has no
|
|
# discovery step, so declare that same required observation explicitly.
|
|
initial_capture = client.worker_reference.capture
|
|
|
|
def capture_with_search_semantics(task_id):
|
|
value = initial_capture(task_id)
|
|
next(element for element in value["elements"] if element["selector"] == '[data-testid="search-input"]')["search_input"] = True
|
|
return value
|
|
|
|
client.worker_reference.capture = capture_with_search_semantics
|
|
task, candidate, _ = learned_detail(client)
|
|
endpoint = "/api/learning/tasks/" + task["id"]
|
|
service, worker = client.app_reference.state.learning, client.worker_reference
|
|
original_call = worker.call
|
|
original_steps = deepcopy(service.steps(task["id"]))
|
|
original_candidate = service.store.one("SELECT digest,manifest_json FROM candidates WHERE id=$1", (candidate["id"],))
|
|
original_observations = len(client.get(endpoint).json()["observations"])
|
|
before, batch = worker.capture(task["id"]), worker.capture(task["id"])
|
|
batch["text"] = "本机合成批次已完整取证，后续预算中断"
|
|
attempts = []
|
|
|
|
async def interrupted(method, params):
|
|
if method == "action":
|
|
attempts.append(deepcopy(params))
|
|
if close_ack is True:
|
|
worker.sessions.pop(params["task_id"], None)
|
|
raise WorkerError(
|
|
"批次时间预算已到", "BATCH_BUDGET", before=before,
|
|
batch_evidence=[{"batch_index": 1, "added_count": 1, "duplicate_count": 0, "capture": batch}],
|
|
diagnostics={"phase": "execute", "blocked_requests": [], "blocked_request_count": 0},
|
|
session_closed=close_ack,
|
|
)
|
|
return await original_call(method, params)
|
|
|
|
worker.call = interrupted
|
|
response = client.post(endpoint + "/browser/action", json=COMMENTS)
|
|
assert response.status_code == 409, response.text
|
|
failure = response.json()["detail"]
|
|
assert failure["code"] == "BATCH_BUDGET" and failure["evidence_status"] == "recorded"
|
|
assert len(attempts) == 1 and service.steps(task["id"]) == original_steps
|
|
assert len(failure["batch_evidence_ids"]) == 1 and len(failure["failure_observation_ids"]) == 2
|
|
after = client.get(endpoint).json()
|
|
observations = after["observations"][original_observations:]
|
|
assert [value["id"] for value in observations] == failure["failure_observation_ids"]
|
|
assert [value["source"] for value in observations] == ["action_failure", "action_failure_batch"]
|
|
assert observations[1]["text"] == batch["text"]
|
|
assert observations[1]["batch"] == {"batch_index": 1, "added_count": 1, "duplicate_count": 0}
|
|
assert all(client.get(value["screenshot_url"]).status_code == 200 for value in observations)
|
|
assert service.store.one("SELECT digest,manifest_json FROM candidates WHERE id=$1", (candidate["id"],)) == original_candidate
|
|
assert after["browser_readiness"]["ready"] is (close_ack is not True)
|
|
assert after["browser_status"] == ("closed" if close_ack is True else "ready")
|
|
if close_ack is True:
|
|
assert after["browser_readiness"]["next_action"] == "open_browser"
|
|
calls_before = len(worker.calls)
|
|
stopped = client.post(endpoint + "/browser/action", json=COMMENTS)
|
|
assert stopped.status_code == 409 and stopped.json()["detail"]["code"] == "browser_session_required"
|
|
assert worker.calls[calls_before:] == [] and len(attempts) == 1
|
|
assert service.steps(task["id"]) == original_steps
|
|
assert len(client.get(endpoint).json()["observations"]) == len(after["observations"])
|
|
|
|
|
|
@pytest.mark.parametrize("close_ack", [True, "true", False], ids=["closed", "string-is-not-ack", "not-closed"])
|
|
def test_batch_close_ack_survives_rpc_only_as_actual_boolean(tmp_path, close_ack):
|
|
payload = {
|
|
"code": "BATCH_BUDGET", "message": "合成批次预算失败", "session_closed": close_ack,
|
|
"before": {"url": "http://127.0.0.1/fixture", "elements": [], "screenshot_path": str(tmp_path / "before.png")},
|
|
"batch_evidence": [{"batch_index": 1, "added_count": 1, "duplicate_count": 0, "capture": {
|
|
"url": "http://127.0.0.1/fixture", "elements": [], "screenshot_path": str(tmp_path / "batch.png"),
|
|
}}],
|
|
"diagnostics": {"phase": "execute", "blocked_requests": [], "blocked_request_count": 0},
|
|
}
|
|
|
|
class Pipe:
|
|
def write(self, value):
|
|
self.request = json.loads(value)
|
|
|
|
async def drain(self):
|
|
pass
|
|
|
|
async def readline(self):
|
|
return json.dumps({"id": self.request["id"], "error": payload}).encode()
|
|
|
|
class Process:
|
|
returncode = None
|
|
stdin = stdout = Pipe()
|
|
|
|
async def check():
|
|
worker = BrowserWorker(tmp_path, tmp_path)
|
|
worker.process = Process()
|
|
with pytest.raises(WorkerError) as failure:
|
|
await worker.call("action", {"task_id": "batch-fixture"})
|
|
error = failure.value
|
|
assert error.code == "BATCH_BUDGET" and error.session_closed is (close_ack is True)
|
|
assert error.before["screenshot_path"].endswith("before.png") and error.after is None
|
|
assert error.batch_evidence[0]["capture"]["screenshot_path"].endswith("batch.png")
|
|
assert error.diagnostics["phase"] == "execute"
|
|
|
|
asyncio.run(check())
|