"""Pure integration of real operation facts with the v0.3 proof boundary. Only the persistence surface is replaced. Real spec normalization, preparation, dependency checking and deterministic projections stay in use. """ from copy import deepcopy from contextlib import contextmanager from datetime import datetime, timezone import json from types import SimpleNamespace import pytest from backend.prospecting import knowledge, verification TASK = {"id": "operation_fixture", "inputs": {"intent": "discover"}} class MemoryKnowledge: def __init__(self): self.methods = {} self.events = {} self.calls = [] def register_method(self, spec, task_id=""): normalized = knowledge.normalize_spec(spec) method_id = knowledge._digest([normalized["key"], normalized["version"]], "method_") method = {"id": method_id, **normalized} if method_id in self.methods: assert self.methods[method_id] == method, "same method version changed between facts" self.methods[method_id] = method return deepcopy(method) def record_event(self, method_id, task_id, object_key, outcome, **kwargs): self.calls.append({"method_id": method_id, "task_id": task_id, "object_key": object_key, "outcome": outcome, **deepcopy(kwargs)}) method = self.methods[method_id] prepared = knowledge.prepare_event(method, task_id, object_key, outcome, **kwargs) if prepared["id"] in self.events: assert self.events[prepared["id"]] == prepared, "same receipt changed on replay" return deepcopy(prepared) if prepared["outcome"] == "success" and method["kind"] == "pipeline": assert knowledge._pipeline_proofs_valid(method, prepared, list(self.events.values())), "pipeline lacks actual dependency events" self.events[prepared["id"]] = prepared return deepcopy(prepared) def task_summary(self, task_id): events = [event for event in self.events.values() if event["task_id"] == task_id] methods = {event["method_id"] for event in events} context = {"platform": "xiaohongshu", "page_variant": verification.PAGE_VARIANT, "available_method_ids": list(self.methods)} return {"task_id": task_id, "methods": knowledge.rank_methods([self.methods[mid] for mid in methods], events, context), "events": deepcopy(events), "effective_event_ids": [event["id"] for event in knowledge.effective_events(events)]} @pytest.fixture def memory(monkeypatch): state = MemoryKnowledge() monkeypatch.setattr(knowledge, "register_method", state.register_method) monkeypatch.setattr(knowledge, "record_event", state.record_event) monkeypatch.setattr(knowledge, "task_summary", state.task_summary) return state def fact(capability="save_source", source_kind="note", note="canonical_note_1", index="1", **updates): verifiers = {"save_source": "source_container_quote_and_persisted", "expand_replies": "reply_delta_and_parent", "inspect_profile": "profile_source_header", "export": "file_manifest"} return verification.operation_fact(capability, source_kind, note, f"{TASK['id']}:evidence:before_{index}", f"{TASK['id']}:evidence:after_{index}", verifier=verifiers[capability], at="2026-10-29T10:00:00+08:00", **updates) def synchronize(memory, facts=(), **artifacts): return verification.synchronize(deepcopy(TASK), {"operation_evidence": list(facts), **artifacts}) @pytest.mark.parametrize("capability,kind", [("save_source", "note"), ("save_source", "comment"), ("save_source", "reply"), ("expand_replies", "reply"), ("inspect_profile", "profile"), ("export", "table")]) def test_program_operation_facts_supply_proofs_accepted_by_real_knowledge_boundary(memory, capability, kind): actual = fact(capability, kind) result = synchronize(memory, [actual]) events = list(memory.events.values()) assert len(events) == 1 and events[0]["outcome"] == "success" event = events[0] assert event["provenance"] == "program" assert event["metadata"]["verified"] is True and event["metadata"]["verifier"] == actual["verifier"] assert event["evidence_refs"] == actual["evidence_refs"] assert event["context"]["platform"] == "xiaohongshu" and event["context"]["source_kind"] == kind assert event["context"]["stage"] == ("delivery" if capability == "export" else "search_read") assert event["context"]["page_variant"] == verification.PAGE_VARIANT assert result["knowledge_summary"]["methods"][0]["stats"]["success"] == 1 @pytest.mark.parametrize("discussion_kind", ["comment", "reply"]) def test_same_canonical_note_builds_only_actual_branch_with_real_dependency_event_ids(memory, discussion_kind): note, discussion = fact(), fact(source_kind=discussion_kind, index="2") result = synchronize(memory, [note, discussion]) pipelines = [row for row in memory.methods.values() if row["kind"] == "pipeline"] assert len(pipelines) == 1 and pipelines[0]["branches"] == [discussion_kind] stage = next(event for event in memory.events.values() if event["method_id"] == pipelines[0]["id"]) operation_events = [event for event in memory.events.values() if event["method_id"] != pipelines[0]["id"]] assert stage["metadata"]["dependency_event_ids"] == [event["id"] for event in operation_events] assert set(pipelines[0]["dependencies"]) == {event["method_id"] for event in operation_events} assert stage["metadata"]["terminal_complete"] is True assert stage["metadata"]["executed_branch"] == discussion_kind assert stage["evidence_refs"] == {"pre": note["evidence_refs"]["pre"], "post": discussion["evidence_refs"]["post"]} assert stage["outcome"] == "success" and stage["metadata"]["verifier"] == "same_note_saved_sources" assert stage["object_digest"] == operation_events[0]["object_digest"] == operation_events[1]["object_digest"] assert len(result["knowledge_event_ids"]) == 3 def test_different_note_successes_cannot_be_combined_into_stage_success(memory): synchronize(memory, [fact(note="canonical_note_1"), fact(source_kind="comment", note="canonical_note_2", index="2")]) assert len(memory.events) == 2 assert all(method["kind"] == "operation" for method in memory.methods.values()) assert len({event["object_digest"] for event in memory.events.values()}) == 2 def test_unknown_discussion_fact_cannot_complete_a_stage(memory): synchronize(memory, [fact(), fact(source_kind="reply", outcome="unknown", index="2")]) assert len(memory.events) == 2 assert not any(method["kind"] == "pipeline" for method in memory.methods.values()) assert {event["outcome"] for event in memory.events.values()} == {"success", "unknown"} def test_claimed_success_with_missing_program_proof_does_not_complete_stage(memory): incomplete = fact(source_kind="reply", index="2") incomplete["evidence_refs"].pop("post") synchronize(memory, [fact(), incomplete]) assert len(memory.events) == 2 and not any(method["kind"] == "pipeline" for method in memory.methods.values()) result = next(event for event in memory.events.values() if event["requested_outcome"] == "success" and event["outcome"] == "unknown") assert result["metadata"]["reason_code"] == "program_proof_required" def test_repeated_fact_sync_has_stable_method_and_event_ids_without_inflating_success(memory): facts = [fact(), fact(source_kind="comment", index="2")] first = synchronize(memory, facts) second = synchronize(memory, facts) assert first == second and len(memory.events) == 3 assert all(method["stats"]["success"] == 1 for method in second["knowledge_summary"]["methods"]) def test_existing_operation_and_stage_receipts_do_not_change_when_task_intent_is_revised(memory): artifacts = {"operation_evidence": [fact(), fact(source_kind="comment", index="2")]} first = verification.synchronize(deepcopy(TASK), deepcopy(artifacts)) changed_task = {**TASK, "inputs": {"intent": "message"}} second = verification.synchronize(changed_task, deepcopy(artifacts)) assert first["knowledge_event_ids"] == second["knowledge_event_ids"] assert len(memory.events) == 3 assert all(event["context"]["stage"] == "search_read" for event in memory.events.values()) def test_same_capability_selected_method_is_not_credited_for_program_template_fact(memory): custom = memory.register_method({**verification.method_spec("save_source", "comment"), "key": "model_alternative_save_comment", "origin": "model"}) result = synchronize(memory, [fact(source_kind="comment")], knowledge_methods=[custom], learning_outcomes=[{"entry_id": custom["id"], "outcome": "confirmed", "evidence": "model says saved"}]) canonical_events = [event for event in memory.events.values() if event["method_id"] != custom["id"]] custom_events = [event for event in memory.events.values() if event["method_id"] == custom["id"]] assert len(canonical_events) == 1 and canonical_events[0]["outcome"] == "success" assert all(event["outcome"] in {"unknown", "unused"} for event in custom_events) assert any(event["outcome"] == "unknown" and event["provenance"] == "model" for event in custom_events) projected = next(method for method in result["knowledge_summary"]["methods"] if method["id"] == custom["id"]) assert projected["stats"]["success"] == 0 and projected["stats"]["contradiction"] == 0 @pytest.mark.parametrize("reported,expected", [("confirmed", "unknown"), ("contradicted", "unknown"), ("unused", "unused")]) def test_model_report_without_operation_fact_is_not_a_technical_verdict(memory, reported, expected): selected = memory.register_method({**verification.method_spec("save_source", "reply"), "key": "model_reply_method", "origin": "model"}) result = synchronize(memory, knowledge_methods=[selected], learning_outcomes=[{"entry_id": selected["id"], "outcome": reported, "evidence": "model report only"}]) reports = [event for event in memory.events.values() if event["provenance"] == "model"] assert reports and all(event["outcome"] in {"unknown", "unused"} for event in reports) assert any(event["outcome"] == expected for event in reports) assert all(method["stats"]["success"] == 0 for method in result["knowledge_summary"]["methods"]) def test_legacy_saved_sources_without_operation_fact_do_not_backfill_historical_success(memory): historical = {"sources": [{"source_id": "old_note", "source_type": "note", "note_id": "old_note", "evidence": "current_visible_dom"}, {"source_id": "old_reply", "source_type": "reply", "note_id": "old_note", "parent_source_id": "old_comment"}], "profiles": [{"source_id": "old_reply", "xiaohongshu_id": "old_saved_value"}]} before = deepcopy(historical) verification.synchronize(deepcopy(TASK), historical) assert memory.methods == {} and memory.events == {} and historical == before def test_operation_fact_identity_ignores_repeated_observation_time_and_hides_raw_object(): original = fact(note="private_actual_profile_identity") replay = verification.operation_fact("save_source", "note", "private_actual_profile_identity", original["evidence_refs"]["pre"], original["evidence_refs"]["post"], verifier=original["verifier"], at="2026-10-30T10:00:00+08:00") assert original["id"] == replay["id"] and original["object_key"] == replay["object_key"] assert "private_actual_profile_identity" not in str(original) def test_persisted_method_projection_can_be_injected_into_a_new_browser_prompt(): from backend.prospecting.contracts import ProspectingBrief from backend.prospecting.prompts import browser_prompt method = {**verification.method_spec("save_source", "comment"), "id": "method_fixture", "created_at": datetime(2026, 10, 29, 10, 0, tzinfo=timezone.utc)} projected = knowledge.evaluate_method(method, [], verification.context_for(TASK)) json.dumps({"knowledge_methods": [projected]}) # Same serializer used by task JSONB artifacts. prompt = browser_prompt(ProspectingBrief(target_customer="fixture", target_candidates=1), [projected]) assert projected["id"] in prompt def test_persisted_event_projection_can_be_saved_in_task_knowledge_summary(): method = {**knowledge.normalize_spec(verification.method_spec("save_source", "note")), "id": "method_fixture"} payload = knowledge.prepare_event(method, TASK["id"], "canonical_note_1", "success", {"pre": "operation_fixture:evidence:before", "post": "operation_fixture:evidence:after"}, verification.context_for(TASK), verified=True, verifier="fixture") projected = knowledge._event_row({"payload": payload, "created_at": datetime(2026, 10, 29, 10, 0, tzinfo=timezone.utc)}) json.dumps({"knowledge_summary": {"events": [projected]}}) def test_business_feedback_timestamps_are_serializable_in_public_receipt_and_summary(monkeypatch): timestamp = datetime(2026, 10, 29, 10, 0, tzinfo=timezone.utc) payload = {"task_id": TASK["id"], "result_version": "result_v1", "goal_version": 1, "feedback": {"rating": "up"}, "source": "user", "attribution": "unproven"} row = {"id": "feedback_fixture", "payload": payload, "created_at": timestamp} @contextmanager def connection(): yield object() monkeypatch.setattr(knowledge.db, "get_conn", connection) monkeypatch.setattr(knowledge.db, "_execute", lambda *_args, **_kwargs: 1) monkeypatch.setattr(knowledge.db, "_fetchone", lambda *_args, **_kwargs: row) monkeypatch.setattr(knowledge.db, "_fetchall", lambda _conn, query, _params: [row] if "prospecting_business_feedback" in query else []) receipt = knowledge.record_business_feedback(TASK["id"], "result_v1", {"rating": "up"}, goal_version=1) summary = knowledge.task_summary(TASK["id"]) assert receipt["created_at"] == timestamp.isoformat() assert summary["business_feedback"][0]["created_at"] == timestamp.isoformat() json.dumps({"receipt": receipt, "knowledge_summary": summary}) @pytest.mark.parametrize("intent,stage", [("export", "delivery"), ("follow", "outreach"), ("message", "outreach"), ("discover", "search_read"), ("inspect", "search_read")]) def test_knowledge_default_context_matches_actual_stage(intent, stage): assert verification.context_for({"inputs": {"intent": intent}})["stage"] == stage assert verification.context_for({"inputs": {"intent": intent}}, stage="search_read")["stage"] == "search_read" def test_fact_trace_is_copied_when_created_not_linked_to_mutable_run_state(): trace = {"run_id": "actual_run_1", "goal_version": 3, "action_id": "actual_action_1"} actual = fact(trace=trace) trace.update(run_id="actual_run_2", goal_version=4) assert actual["details"]["trace"] == {"run_id": "actual_run_1", "goal_version": 3, "action_id": "actual_action_1"} def test_actual_trace_and_pipeline_terminal_trace_survive_goal_revision_and_replay(memory): note = fact(trace={"run_id": "note_run", "goal_version": 2, "action_id": "note_action"}) discussion = fact(source_kind="reply", index="2", trace={"run_id": "discussion_run", "goal_version": 3, "action_id": "discussion_action"}) artifacts = {"operation_evidence": [note, discussion]} first = verification.synchronize({**TASK, "inputs": {"intent": "discover", "_goal_version": 3}}, deepcopy(artifacts)) before = deepcopy(memory.events) replay = verification.synchronize({**TASK, "inputs": {"intent": "message", "_goal_version": 99}}, deepcopy(artifacts)) assert first["knowledge_event_ids"] == replay["knowledge_event_ids"] and memory.events == before for event in memory.events.values(): method = memory.methods[event["method_id"]] expected = note["details"]["trace"] if method["kind"] == "operation" and method["source_kind"] == "note" else discussion["details"]["trace"] assert {key: event["metadata"][key] for key in expected} == expected pipeline = next(event for event in memory.events.values() if memory.methods[event["method_id"]]["kind"] == "pipeline") assert pipeline["metadata"]["goal_version"] == 3 and pipeline["metadata"]["run_id"] == "discussion_run" def test_legacy_fact_without_trace_stays_unattributed_despite_current_goal_and_run(memory): artifacts = {"operation_evidence": [fact(), fact(source_kind="comment", index="2")], "_knowledge_goal_version": 99, "run_id": "latest_run_should_not_backfill"} first = verification.synchronize({**TASK, "inputs": {"_goal_version": 99}}, deepcopy(artifacts)) replay = verification.synchronize({**TASK, "inputs": {"_goal_version": 100}}, deepcopy(artifacts)) assert first["knowledge_event_ids"] == replay["knowledge_event_ids"] assert len(memory.events) == 3 for event in memory.events.values(): assert not {"run_id", "goal_version", "action_id"} & event["metadata"].keys() def test_tool_context_is_snapshot_only_and_cleared_without_enriching_old_fact(): from backend.browser import tools task_id = "operation_trace_fixture" tools.clear_task_context(task_id) try: tools.set_prospecting_context(task_id) tools.set_prospecting_run_context(task_id, "run_one", goal_version=3) tools._operation_evidence(task_id, "save_source", "note", "note_fixture", "pre_one", "post_one", verifier="fixture") original = deepcopy(tools.get_task_artifacts(task_id)["operation_evidence"][0]) assert original["details"]["trace"] == {"run_id": "run_one", "goal_version": 3, "action_id": original["id"]} tools.set_prospecting_run_context(task_id, "run_two") tools._operation_evidence(task_id, "save_source", "note", "note_fixture", "pre_two", "post_two", verifier="fixture") saved = tools.get_task_artifacts(task_id)["operation_evidence"] assert saved[0] == original assert saved[1]["details"]["trace"] == {"run_id": "run_two", "action_id": saved[1]["id"]} tools.clear_task_context(task_id) tools.set_prospecting_context(task_id, artifacts={"operation_evidence": [original]}) tools._operation_evidence(task_id, "save_source", "note", "note_fixture", "pre_three", "post_three", verifier="fixture") assert "trace" not in tools.get_task_artifacts(task_id)["operation_evidence"][1]["details"] old = fact() tools.set_prospecting_context(task_id, artifacts={"operation_evidence": [old]}) tools.set_prospecting_run_context(task_id, "new_run", goal_version=9) tools._operation_evidence(task_id, old["capability"], old["source_kind"], "canonical_note_1", old["evidence_refs"]["pre"], old["evidence_refs"]["post"], verifier=old["verifier"]) assert tools.get_task_artifacts(task_id)["operation_evidence"] == [old] assert "trace" not in old["details"] finally: tools.clear_task_context(task_id) async def test_real_runner_run_ids_and_supplied_goal_versions_are_captured_per_action(tmp_path, monkeypatch): from backend.browser import runner, tools task_id = "operation_run_trace_fixture" monkeypatch.setattr(runner, "task_doc_dir", lambda _task_id: tmp_path) monkeypatch.setattr(runner, "get_llm_config", lambda: {"provider": "fixture", "model": "fixture"}) monkeypatch.setattr(runner, "append_progress", lambda *_args, **_kwargs: None) class Agent: def __init__(self): self.state = SimpleNamespace(n_steps=1) self.count = 0 self.history = SimpleNamespace(history=[], number_of_steps=lambda: self.count, total_duration_seconds=lambda: 0, is_done=lambda: True, is_successful=lambda: True, urls=lambda: [], action_names=lambda: [], errors=lambda: [], final_result=lambda: "fixture_complete") def add_new_task(self, _message): pass async def run(self, max_steps, on_step_start=None, on_step_end=None): await on_step_start(self) self.count += 1 tools._operation_evidence(task_id, "save_source", "note", "same_note", f"before_{self.count}", f"after_{self.count}", verifier="fixture") self.state.n_steps += 1 if on_step_end: await on_step_end(self) return self.history tools.clear_task_context(task_id) tools.set_prospecting_context(task_id) agent = Agent() monkeypatch.setitem(runner._live, task_id, runner.LiveSession(agent, None, tool_profile="prospecting")) try: await runner.run_browser_step(task_id=task_id, agent_task="fixture", profile_path="fixture", allowed_domains=[], continuation="first", max_steps=1, keep_session=True, tool_profile="prospecting", prospecting_artifacts={"_knowledge_goal_version": 3}) first_run_id = json.loads((tmp_path / "browser-budget.json").read_text())["last_run_id"] first = deepcopy(tools.get_task_artifacts(task_id)["operation_evidence"][0]) await runner.run_browser_step(task_id=task_id, agent_task="fixture", profile_path="fixture", allowed_domains=[], continuation="second", max_steps=1, keep_session=True, tool_profile="prospecting", prospecting_artifacts={"_knowledge_goal_version": 4}) second_run_id = json.loads((tmp_path / "browser-budget.json").read_text())["last_run_id"] saved = tools.get_task_artifacts(task_id)["operation_evidence"] assert first_run_id != second_run_id and saved[0] == first assert saved[0]["details"]["trace"] == {"run_id": first_run_id, "goal_version": 3, "action_id": saved[0]["id"]} assert saved[1]["details"]["trace"] == {"run_id": second_run_id, "goal_version": 4, "action_id": saved[1]["id"]} finally: runner._live.pop(task_id, None) tools.clear_task_context(task_id) def test_export_trace_uses_manifest_goal_and_export_id_without_browser_run(memory, tmp_path, monkeypatch): from backend.prospecting import workflows path = tmp_path / "fixture.csv" path.write_text("nickname\nfixture\n") metadata = {"result_version": "result_fixture", "goal_version": 3, "count": 1} monkeypatch.setattr(workflows, "export_path", lambda _task, _export_id: (path, metadata)) monkeypatch.setattr(workflows, "result_version", lambda _task: "result_fixture") entry = {"id": "export_fixture", "generated_at": "2026-10-29T10:00:00+08:00"} task = {**TASK, "inputs": {"intent": "export", "_goal_version": 99}, "artifacts": {}} artifacts = verification.export_fact(task, entry) actual = artifacts["operation_evidence"][0] assert actual["details"]["trace"] == {"goal_version": 3, "action_id": entry["id"]} verification.synchronize(task, artifacts) event = next(iter(memory.events.values())) assert event["metadata"]["goal_version"] == 3 and event["metadata"]["action_id"] == entry["id"] assert "run_id" not in event["metadata"] replay = verification.export_fact({**task, "inputs": {"_goal_version": 100}, "artifacts": artifacts}, entry) assert replay == artifacts