assistant/tests/test_baseline_measurement_p1.py

"""Measurement must preserve gaps and separate proposals from invoked tools."""
import importlib.util
import json
from pathlib import Path


spec = importlib.util.spec_from_file_location("baseline_measurement_p1", Path(__file__).resolve().parents[1] / "scripts/measure-prospecting-baseline.py")
subject = importlib.util.module_from_spec(spec)
spec.loader.exec_module(subject)


def event(kind, cid="a" * 32, phase="browser", **extra):
    return {"event_id": cid + ":" + kind, "kind": kind, "call_id": cid, "phase": phase,
            "model": "fixture", "at": "2026-10-30T01:00:00+08:00", **extra}


ORIGIN = {"event_id": "task-start", "kind": "task_start", "at": "2026-10-30T01:00:00+08:00"}
USAGE = {"input_tokens": 100, "output_tokens": 10, "total_tokens": 110, "cache_hit_tokens": 50, "cache_miss_tokens": 50}


def test_unknown_call_keeps_known_sum_but_complete_totals_and_per_candidate_denominator_cannot_hide_it():
    events = [ORIGIN, event("started"), event("response", usage=USAGE),
              event("finished", status="succeeded"), event("started", cid="b" * 32),
              event("finished", cid="b" * 32, status="connection_error")]
    report = subject.usage_report(events, False)
    assert report["total"]["calls_total"] == 2
    assert report["total"]["known_tokens"]["total_tokens"] == 110
    assert report["total"]["tokens_total"]["total_tokens"] is None
    assert report["total"]["unknown_usage_requests"] == 1
    assert report["phases"]["analysis"]["tokens_total"]["total_tokens"] == 0
    assert subject.usage_report(events[1:], False)["total"]["calls_total"] is None


def test_incremental_analysis_is_grouped_and_review_is_not_hidden_from_total_cost():
    events = [ORIGIN, event("response", phase="analysis_incremental", usage=USAGE),
              event("response", cid="b" * 32, phase="review", usage=USAGE)]
    report = subject.usage_report(events, False)
    assert report["phases"]["analysis"]["calls_total"] == 1
    assert report["phases"]["review"]["calls_total"] == 1
    assert report["total"]["tokens_total"]["total_tokens"] == 220


def test_zero_delivery_missing_ledgers_and_visits_are_not_filled_with_success_or_zero_cost(tmp_path):
    note_id = "a" * 24
    # A supplied but unopened link in the goal is not a page observation.
    (tmp_path / "progress.md").write_text("- **目标**：检查 https://www.xiaohongshu.com/explore/" + "c" * 24)
    task = {"id": "task_fixture", "status": "blocked", "created_at": "2026-10-30 01:00:00",
            "updated_at": "2026-10-30 01:00:30", "inputs": {"intent": "discover", "target_candidates": 10},
            "artifacts": {"stop_reason": "no_progress", "browser_history": {"urls": ["https://www.xiaohongshu.com/explore/" + note_id]}}}
    report = subject.build_baseline(task, tmp_path)
    assert report["reading"]["notes_visited_observed"] == 1
    assert report["reading"]["notes_with_saved_quotes"] == 0
    assert report["candidates"]["delivered_stable_identity"] == 0
    assert report["per_candidate"]["calls"] is None
    assert report["per_candidate"]["tokens"] is None
    assert report["per_candidate"]["wall_clock_seconds"] is None
    assert report["llm"]["total"]["calls_total"] is None
    assert report["browser"]["total_steps_attempted"] is None
    assert report["browser"]["executed_tools"]["invoked_total"] is None
    assert report["timing"]["wall_clock_seconds"] == 30
    assert report["stop"]["reason"] == "no_progress"
    assert note_id not in json.dumps(report)


def test_actual_boundary_ledger_counts_invoked_once_and_keeps_blocked_separate(tmp_path):
    path = tmp_path / "browser-actions.jsonl"
    base = {"version": 1, "kind": "browser_action", "at": "2026-10-30T01:00:00Z"}
    rows = [{**base, "action_id": "a" * 32, "action_name": "save_source", "status": status}
            for status in ("invoked", "returned", "returned")]
    rows.append({**base, "action_id": "b" * 32, "action_name": "click", "status": "blocked"})
    path.write_text("\n".join(json.dumps(row) for row in rows))
    report = subject.executed_tools(path)
    assert report["invoked"] == {"save_source": 1}
    assert report["invoked_total"] == 1
    assert report["status_counts"]["blocked"] == {"click": 1}
    assert report["pending_calls"] == 0
    proposals = subject.step_report("### ℹ️ [2026-10-30 01:00:00] 步骤 1\n\n- **详情**：Agent 累计步骤：1 | 目标：fixture | 动作：click、save_source | 页面：https://example.test\n", {}, {}, 0)
    assert proposals["proposed_tool_call_counts"] == {"click": 1, "save_source": 1}


def test_candidate_counts_deduplicate_stable_identity_and_keep_unknown_separate(tmp_path):
    sources = [{"source_id": sid, "source_type": "comment", "note_id": "a" * 24,
                "content": "准备装修", **({"author_profile_url": "https://www.xiaohongshu.com/user/profile/" + "b" * 24} if sid != "unknown" else {})}
               for sid in ("one", "same", "unknown")]
    task = {"id": "task_fixture", "inputs": {"intent": "discover", "target_candidates": 10},
            "artifacts": {"sources": sources, "leads": [{"source_id": row["source_id"], "decision": "keep", "quote": "准备装修"} for row in sources]}}
    report = subject.build_baseline(task, tmp_path)
    assert report["candidates"]["delivered_stable_identity"] == 1
    assert report["candidates"]["unknown_identity_groups"] == 1
    assert report["candidates"]["remaining"] == 9
    assert report["reading"]["saved_comments"] == 3
    assert report["reading"]["notes_with_saved_quotes"] == 1


def test_request_contract_requires_matching_provider_response_for_real_request_evidence(tmp_path):
    path = tmp_path / "model-request-contracts.jsonl"
    path.write_text(json.dumps({"kind": "request_prepared", "call_id": "a" * 32, "action_names": ["save_source", "scroll"]}) + "\n")
    assert subject.request_contracts(path, [ORIGIN])["requests_with_response"] == 0
    report = subject.request_contracts(path, [ORIGIN, event("response", usage=USAGE)])
    assert report["requests_with_response"] == 1
    assert report["disabled_tools_exposed"] == []