{escape(reason)}
+This link is single-use and expires 24 hours after it was issued. +Ask the requester to re-trigger the approval if you still need to act on it.
Run: {payload.get("rid","")}
-Step: {payload.get("step","")}
-Expires: {expires_at}
+ # Enrich with the workflow name when a run store is wired (fail-soft). + workflow_name = "" + store = getattr(app_state, "run_store", None) + if store is not None and run_id: + try: + run = await store.get_run(run_id) + if run is not None: + workflow_name = str(run.workflow) + except Exception: + workflow_name = "" + + workflow_line = ( + f'Workflow: {escape(workflow_name)}
' if workflow_name else "" + ) + body = f"""Run: {escape(run_id)}
+Step: {escape(step)}
+Expires: {escape(expires_at)}
Reason recorded: {escape(reason)}
" if reason else "" + body = f"""The workflow step was {escape(outcome)}.
+Run: {escape(result.run_id)}
+Gate: {escape(result.gate_request_id)}
+ {reason_line} +You can close this window — the workflow continues automatically.
""" + return _page("Decision recorded", body) diff --git a/tests/test_e2e_workflows.py b/tests/test_e2e_workflows.py index 01a9b7f..590fab8 100644 --- a/tests/test_e2e_workflows.py +++ b/tests/test_e2e_workflows.py @@ -257,3 +257,26 @@ async def test_e2e_05_duplicate_lead_produces_single_run() -> None: assert r1["run_id"] == r2["run_id"], "Duplicate trigger must return the same run" assert len(run_store._runs) == 1, "Only one run should exist" + assert r1["is_new"] is True, "First submission must report a new run" + assert r2["is_new"] is False, "Duplicate submission must report the existing run" + + +async def test_e2e_06_workflow_run_reports_is_new() -> None: + """workflow.run reports is_new=False when the idem key dedupes to an existing run.""" + from cam.mcp_server.workflow_tools import WorkflowRunInput, tool_workflow_run + + services = _make_services() + register_intake_workflow(services) + + run_store = InMemoryRunStore() + inp = WorkflowRunInput( + workflow="intake", + context={"lead": _make_lead().model_dump(mode="json")}, + idem_key="e2e-06-idem-001", + ) + out1 = await tool_workflow_run(inp, run_store) + out2 = await tool_workflow_run(inp, run_store) + + assert out1.is_new is True + assert out2.is_new is False + assert out1.run_id == out2.run_id diff --git a/tests/test_intake.py b/tests/test_intake.py index f206154..c50676d 100644 --- a/tests/test_intake.py +++ b/tests/test_intake.py @@ -361,6 +361,8 @@ async def test_duplicate_lead_same_run() -> None: r1 = await tool_intake_run(lead, store, idempotency_key="idem-abc") r2 = await tool_intake_run(lead, store, idempotency_key="idem-abc") assert r1["run_id"] == r2["run_id"] + assert r1["is_new"] is True + assert r2["is_new"] is False # ────────────────────────────────────────────────────────────── diff --git a/tests/test_orchestrator_gates.py b/tests/test_orchestrator_gates.py index 583f4d9..b8dbd69 100644 --- a/tests/test_orchestrator_gates.py +++ b/tests/test_orchestrator_gates.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import Any + import pytest from cam.core.orchestrator.dsl import GateConfig, StepContext, clear_registry, workflow @@ -130,3 +132,29 @@ def test_token_verify_wrong_key() -> None: raw_token, _ = issue_token("gid", "rid", "GATE:step", "mcp", KEY) _, err = verify_token(raw_token, b"wrong-key-32-bytes-padded-00000") assert err == "tampered" + + +# h. Approver reason is recorded on the decision and in the audit stream +async def test_reason_recorded_on_decision() -> None: + store, run_id, raw_token, _ = await _build_gated_run("wf_reason") + audits: list[dict[str, Any]] = [] + + async def _audit(**kwargs: Any) -> None: + audits.append(kwargs) + + await resolve_gate( + raw_token, "approve", "attorney", "mcp", KEY, store, + audit_fn=_audit, reason="Reviewed draft; ready to send", + ) + decision = store.all_decisions()[0] + assert decision.reason == "Reviewed draft; ready to send" + approved = [a for a in audits if a.get("action") == "gate.approved"] + assert approved, "gate.approved must be audited" + assert approved[0]["inputs"]["reason"] == "Reviewed draft; ready to send" + + +# i. Omitted reason stays None (backwards compatible) +async def test_reason_defaults_to_none() -> None: + store, run_id, raw_token, _ = await _build_gated_run("wf_no_reason") + await resolve_gate(raw_token, "approve", "attorney", "mcp", KEY, store) + assert store.all_decisions()[0].reason is None diff --git a/tests/test_sidecar.py b/tests/test_sidecar.py index c53df20..35168dc 100644 --- a/tests/test_sidecar.py +++ b/tests/test_sidecar.py @@ -5,7 +5,7 @@ import hashlib import hmac import json -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta import pytest from fastapi.testclient import TestClient @@ -13,6 +13,14 @@ from cam.connectors.reference import ReferenceCaseConnector from cam.connectors.registry import clear_registry, register_connector from cam.connectors.webhook.pipeline import reference_normaliser +from cam.core.orchestrator.gates import issue_token +from cam.core.orchestrator.states import ( + GateRequest, + RunStatus, + TriggerRef, + WorkflowRun, +) +from cam.core.orchestrator.store import InMemoryRunStore from cam.sidecar.main import create_app NOW = datetime(2026, 6, 1, 12, 0, 0, tzinfo=UTC) @@ -53,6 +61,12 @@ def _fake_sink_calls_append(self, e): self.calls.append(e) sink = _FakeSink() redis = _FakeRedis() + # Simulate an authenticated approver session (production wires SSO here). + @app.middleware("http") + async def _fake_auth(request, call_next): # type: ignore[no-untyped-def] + request.state.user_identity = "attorney" + return await call_next(request) + app.state.webhook_secrets = {"reference": WEBHOOK_SECRET} app.state.redis_client = redis app.state.trigger_sink = sink @@ -143,7 +157,6 @@ def test_webhook_unknown_event_type_202(client): def _make_raw_token(gate_id: str, run_id: str, step: str = "GATE:approve") -> str: - from cam.core.orchestrator.gates import issue_token raw_token, _ = issue_token(gate_id, run_id, step, "web", SIGNING_KEY) return raw_token @@ -177,3 +190,79 @@ def test_approval_post_bad_decision_400(client): token = _make_raw_token("gate-1", "run-1") r = client.post(f"/approvals/{token}", data={"decision": "maybe"}) assert r.status_code in (400, 422, 503), f"Unexpected status {r.status_code}" + + +# --------------------------------------------------------------------------- +# Approval UI — human-readable pages, reason capture, workflow context +# --------------------------------------------------------------------------- + + +async def _seed_gated_run( + app, run_id: str = "run-web-1", gate_id: str = "gate-web-1", ttl_seconds: int = 86400 +) -> tuple[InMemoryRunStore, str]: + """Create a run + pending gate + token and wire the store into the app.""" + store = InMemoryRunStore() + now = datetime.now(tz=UTC) + store._runs[run_id] = WorkflowRun( + id=run_id, + workflow="intake", + workflow_version=1, + status=RunStatus.AWAITING_APPROVAL, + trigger=TriggerRef(kind="agent", source="test"), + created_at=now, + updated_at=now, + ) + await store.create_gate_request( + GateRequest( + id=gate_id, + run_id=run_id, + step="GATE:human_review", + required_role="attorney", + status="pending", + created_at=now, + expires_at=now + timedelta(seconds=ttl_seconds), + ) + ) + raw_token, token_record = issue_token(gate_id, run_id, "GATE:human_review", "web", + SIGNING_KEY, ttl_seconds=ttl_seconds) + await store.save_token(token_record) + app.state.run_store = store + return store, raw_token + + +async def test_approval_page_shows_workflow_and_reason_field(client): + _, raw_token = await _seed_gated_run(client.app) + r = client.get(f"/approvals/{raw_token}") + assert r.status_code == 200 + assert "intake" in r.text, "Page should show the workflow being approved" + assert 'name="reason"' in r.text, "Page should offer a reason field" + + +async def test_approval_page_invalid_token_is_html_401(client): + token = _make_raw_token("gate-1", "run-1") + tampered = token[:-4] + "XXXX" + r = client.get(f"/approvals/{tampered}") + assert r.status_code == 401 + assert "