import asyncio import base64 import json from pathlib import Path from unittest.mock import patch from a2a_pack import LocalRunContext, PlatformUserAuth from a2a_pack.cli.local import load_local_project from agent import ( BrowserDocument, ContractClockStudioV1, extract_timeline, production_receipt_id, ) ROOT = Path(__file__).resolve().parents[1] SUCCESS_TEXT = "This Agreement renews automatically on 2026-12-31 unless either party gives at least 60 days written notice. Invoices are due 30 days after receipt." FAILURE_TEXT = "This Agreement renews sometime next spring unless notice is given well in advance." def ctx(): return LocalRunContext( auth=PlatformUserAuth(sub="user-123", user_id=123, email="user@example.test", scopes=["agent:invoke"]), caller="tester", ) def test_full_stack_product_contract(): manifest = (ROOT / "a2a.yaml").read_text(encoding="utf-8") source = (ROOT / "agent.py").read_text(encoding="utf-8") frontend = (ROOT / "frontend" / "src" / "App.jsx").read_text(encoding="utf-8") assert "frontend:" in manifest and "mount: /app" in manifest assert "public: false" in manifest assert "resources:" in manifest and "databases:" in manifest assert "migrations:" in manifest and "db/migrations" in manifest assert "PlatformUserAuth" in source assert "AgentPlatformResources" in source and "AgentDatabase(" in source assert "Skill runner" not in frontend assert "ContractClock" in frontend and "analyze_contract" in frontend assert "A2A_LITELLM" not in source and "OPENAI_API_KEY" not in source migrations = list((ROOT / "db" / "migrations").glob("*.sql")) assert migrations and all(path.read_text(encoding="utf-8").strip() for path in migrations) agent_cls = load_local_project(ROOT).agent_cls() card = agent_cls().card() skill_names = {skill.name for skill in card.skills} assert {"analyze_contract", "get_contract_deadlines", "analyze_contract_browser_upload", "analyze_contract_file_upload"}.issubset(skill_names) analyze_schema = next(skill.input_schema for skill in card.skills if skill.name == "analyze_contract") assert set(analyze_schema["required"]) == {"contract_id", "text", "title"} upload_schema = next(skill.input_schema for skill in card.skills if skill.name == "analyze_contract_file_upload") assert upload_schema["properties"]["document"]["x-a2a-file-upload"]["max_bytes"] == 262144 databases = card.runtime.platform_resources.databases assert databases and databases[0].scope == "user" assert databases[0].migrations is not None assert databases[0].migrations.path == "db/migrations" assert card.runtime.pricing.caller_pays_llm is False assert agent_cls.workspace_access.enabled is True def test_extractor_success_fixture_exact_dates(): timeline = extract_timeline(SUCCESS_TEXT) compact = [{"kind": item.kind, "date": item.date} for item in timeline.deadlines] assert {"kind": "renewal", "date": "2026-12-31"} in compact assert {"kind": "notice", "date": "2026-11-01"} in compact assert not any(item["date"] == "2026-01-01" for item in compact) def test_extractor_rejects_ambiguous_fixture(): try: extract_timeline(FAILURE_TEXT) except Exception as exc: assert getattr(exc, "code", None) == "ambiguous_date" else: raise AssertionError("ambiguous fixture should fail") def test_acceptance_success_reload_failure_and_receipt_persistence(): agent = ContractClockStudioV1() saved = {} receipts = [] def fake_persist_contract(tenant, contract_id, title, payload, receipt_id): assert tenant == "user:123" saved[(tenant, contract_id)] = { "payload": payload, "receipt_id": receipt_id, "updated_at": "2026-01-01T00:00:00+00:00", } def fake_load_contract(tenant, contract_id): return saved.get((tenant, contract_id)) def fake_persist_receipt(tenant, receipt_id, skill_name, contract_id, payload, status): receipts.append((tenant, receipt_id, skill_name, contract_id, status)) with patch("agent.persist_contract", side_effect=fake_persist_contract), patch("agent.load_contract", side_effect=fake_load_contract), patch("agent.persist_receipt", side_effect=fake_persist_receipt): result = asyncio.run(agent.analyze_contract(ctx(), contract_id="studio-contract-v1", text=SUCCESS_TEXT, title="Studio Renewal Agreement")) assert result.ok is True assert result.persisted is True assert result.contract_id == "studio-contract-v1" assert [{"kind": d.kind, "date": d.date} for d in result.deadlines[:2]] == [ {"kind": "renewal", "date": "2026-12-31"}, {"kind": "notice", "date": "2026-11-01"}, ] assert result.receipt_id assert receipts and receipts[0][2] == "analyze_contract" reload = asyncio.run(agent.get_contract_deadlines(ctx(), contract_id="studio-contract-v1")) assert reload.ok is True assert reload.receipt_id == result.receipt_id assert [{"kind": d.kind, "date": d.date} for d in reload.deadlines[:2]] == [ {"kind": "renewal", "date": "2026-12-31"}, {"kind": "notice", "date": "2026-11-01"}, ] failure = asyncio.run(agent.analyze_contract(ctx(), contract_id="studio-contract-invalid", text=FAILURE_TEXT, title="Ambiguous Agreement")) assert failure.ok is False assert failure.code == "ambiguous_date" def test_browser_upload_bridge_bounds_and_reuses_workflow(): agent = ContractClockStudioV1() encoded = base64.b64encode(SUCCESS_TEXT.encode("utf-8")).decode("ascii") with patch("agent.persist_contract", return_value=None), patch("agent.persist_receipt", return_value=None): result = asyncio.run( agent.analyze_contract_browser_upload( ctx(), contract_id="studio-contract-v1", title="Studio Renewal Agreement", document=BrowserDocument(filename="contract.txt", media_type="text/plain", data_base64=encoded), ) ) assert result.ok is True bad = asyncio.run( agent.analyze_contract_browser_upload( ctx(), contract_id="studio-contract-v1", title="Studio Renewal Agreement", document=BrowserDocument(filename="contract.txt", media_type="text/plain", data_base64="not base64***"), ) ) assert bad.ok is False and bad.code == "invalid_base64" def test_receipt_id_is_deterministic_without_secrets(): payload = {"contract_id": "studio-contract-v1", "deadlines": [{"kind": "renewal", "date": "2026-12-31"}]} receipt = production_receipt_id("user:123", "analyze_contract", payload) assert receipt == production_receipt_id("user:123", "analyze_contract", payload) assert receipt.startswith("ccr_") assert "DATABASE" not in json.dumps(payload)