From 0530ac79318ed96aadef3063f5078494985be5f8 Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Tue, 14 Jul 2026 10:33:06 -0700 Subject: [PATCH 1/3] INT-O1: enforce entrypoint review validation and provenance --- .../dashboard/dist/index.js | 4 + .../dashboard/plugin_api.py | 21 +++- src/hermes_workflows/approvals.py | 75 ++++++++++++ .../hermes_plugin_approvals.py | 57 +++++++-- tests/test_approval_provenance.py | 46 +++++++ tests/test_dashboard_plugin.py | 87 +++++++++++-- tests/test_hermes_plugin_approvals.py | 114 +++++++++++++++++- 7 files changed, 382 insertions(+), 22 deletions(-) diff --git a/plugins/hermes-workflows-approvals/dashboard/dist/index.js b/plugins/hermes-workflows-approvals/dashboard/dist/index.js index 2e9772a..a44ad1d 100644 --- a/plugins/hermes-workflows-approvals/dashboard/dist/index.js +++ b/plugins/hermes-workflows-approvals/dashboard/dist/index.js @@ -667,6 +667,10 @@ const feedbackAction = (actions || []).find(function (item) { return item.requires_feedback && item.value !== "approve"; }); if (feedbackAction) selectedAction = feedbackAction.value; } + if (String(selectedAction || "").toLowerCase().replace(/-/g, "_") === "request_changes" && !feedbackText && !editedOutputText) { + setUi(Object.assign({}, ui, { error: "request_changes requires nonblank feedback or valid edited_output" })); + return; + } const payload = { action: selectedAction, feedback: feedbackText || undefined }; const editable = (step.input_surface || {}).editable_output || {}; const editField = editable.field || "edited_output"; diff --git a/plugins/hermes-workflows-approvals/dashboard/plugin_api.py b/plugins/hermes-workflows-approvals/dashboard/plugin_api.py index 79ab8ec..309ca27 100644 --- a/plugins/hermes-workflows-approvals/dashboard/plugin_api.py +++ b/plugins/hermes-workflows-approvals/dashboard/plugin_api.py @@ -13,14 +13,17 @@ from typing import Any from hermes_workflows import ApprovalDecisionInput, WorkflowEngine +from hermes_workflows.approvals import strip_client_controlled_provenance, validate_revision_response from hermes_workflows.artifacts import artifact_descriptor, workflow_source_preview from hermes_workflows.hermes_plugin_approvals import ( _configured_dbs, _next_step_for_receipt, _redact, _receipt_to_payload, + _revision_schema_for_response, approval_view_to_dict, ) +from hermes_workflows.revision_validation import RevisionActionValidationError from hermes_workflows.workflow_loading import load_workflow_ref try: # FastAPI is provided by Hermes Agent's dashboard process. @@ -30,8 +33,8 @@ class _FallbackHTTPException(Exception): - def __init__(self, status_code: int, detail: str): - super().__init__(detail) + def __init__(self, status_code: int, detail: Any): + super().__init__(str(detail)) self.status_code = status_code self.detail = detail @@ -2240,7 +2243,7 @@ async def respond_review_request(body: dict[str, Any]) -> dict[str, Any]: raw_payload = body.get("payload") if isinstance(body.get("payload"), dict) else {} if not raw_payload: raise HTTPException(status_code=400, detail="review response payload is required") - payload = {key: value for key, value in raw_payload.items() if key not in {"by", "source"}} + payload = strip_client_controlled_provenance(raw_payload) raw_idempotency_key = str(body.get("idempotency_key") or body.get("event_id") or "").strip() message_id = raw_idempotency_key or f"dashboard:{uuid.uuid4()}" if not message_id.startswith("dashboard:"): @@ -2250,15 +2253,19 @@ def record_and_resume() -> tuple[Any, dict[str, Any]]: _ensure_workflow_project_on_path(db_path) engine = WorkflowEngine(db_path) normalized_payload = _normalize_review_payload_for_dashboard_request(engine, workflow_id, key, payload) + effective_idempotency_key = message_id + if _revision_schema_for_response(engine, workflow_id, key) is not None: + normalized_payload, validated = validate_revision_response(normalized_payload) + effective_idempotency_key = f"revision:{workflow_id}:{key}:{validated.idempotency_key}" receipt = engine.submit_operator_response( workflow_id=workflow_id, key=key, payload=normalized_payload, source={ - "channel": "hermes-dashboard", + "channel": "local-dashboard", "message_id": message_id, }, - idempotency_key=message_id, + idempotency_key=effective_idempotency_key, resume=True, ) post_resume = _status_packet( @@ -2274,6 +2281,8 @@ def record_and_resume() -> tuple[Any, dict[str, Any]]: try: receipt, post_resume = await asyncio.to_thread(record_and_resume) + except RevisionActionValidationError as exc: + raise HTTPException(status_code=400, detail=exc.to_dict()) from exc except Exception as exc: raise HTTPException(status_code=400, detail=f"review response/resume failed: {type(exc).__name__}: {exc}") from exc receipt_payload = _receipt_to_payload(receipt, resume_requested=True) @@ -2307,7 +2316,7 @@ async def decide_approval(body: dict[str, Any]) -> dict[str, Any]: key=key, action=action, source={ - "channel": "hermes-dashboard", + "channel": "local-dashboard", "message_id": message_id, }, note=body.get("note"), diff --git a/src/hermes_workflows/approvals.py b/src/hermes_workflows/approvals.py index 8cd39d9..51b0cda 100644 --- a/src/hermes_workflows/approvals.py +++ b/src/hermes_workflows/approvals.py @@ -3,6 +3,73 @@ from dataclasses import dataclass from typing import Any, Iterator, Mapping +from .provenance import project_response_provenance +from .revision_validation import ValidatedRevisionActionV1, validate_revision_action + + +CLIENT_CONTROLLED_PROVENANCE_FIELDS = frozenset( + { + "actor", + "authenticated_principal", + "by", + "display_label", + "principal", + "provenance", + "response_provenance", + "source", + "user", + } +) + + +def strip_client_controlled_provenance(payload: Mapping[str, Any]) -> dict[str, Any]: + """Remove identity/provenance assertions from an untrusted response body.""" + + if not isinstance(payload, Mapping): + raise TypeError("response payload must be an object") + return {key: value for key, value in payload.items() if key not in CLIENT_CONTROLLED_PROVENANCE_FIELDS} + + +def is_revision_action_schema(schema_descriptor: Any) -> bool: + """Return whether a typed operator request uses the v1 revision action contract.""" + + if not isinstance(schema_descriptor, Mapping): + return False + fields = schema_descriptor.get("fields") + if not isinstance(fields, list): + return False + action = next((field for field in fields if isinstance(field, Mapping) and field.get("name") == "action"), None) + if not isinstance(action, Mapping): + return False + options = action.get("options") if isinstance(action.get("options"), list) else schema_descriptor.get("choices") + field_names = {str(field.get("name")) for field in fields if isinstance(field, Mapping)} + return ( + isinstance(options, list) + and {"approve", "request_changes"}.issubset(options) + and bool({"feedback", "edited_output"} & field_names) + ) + + +def validate_revision_response(payload: Mapping[str, Any]) -> tuple[dict[str, Any], ValidatedRevisionActionV1]: + """Validate and materialize a revision response for durable adapter submission.""" + + validated = validate_revision_action(payload) + return _materialize_json(validated.normalized_payload), validated + + +def response_provenance_for(*, by: str | None, source: Mapping[str, Any] | None) -> dict[str, Any]: + """Truthfully classify an adapter response without upgrading client labels.""" + + return project_response_provenance({"by": by, "source": source}).to_dict() + + +def _materialize_json(value: Any) -> Any: + if isinstance(value, Mapping): + return {str(key): _materialize_json(item) for key, item in value.items()} + if isinstance(value, tuple): + return [_materialize_json(item) for item in value] + return value + @dataclass(frozen=True) class ApprovalView: @@ -60,6 +127,10 @@ def needs_revision(self) -> bool: def feedback(self) -> str | None: return self.direct_feedback or self.reason or self.note or self.message or self.comment + @property + def response_provenance(self) -> dict[str, Any]: + return response_provenance_for(by=self.by, source=self.source) + def to_dict(self) -> dict[str, Any]: data: dict[str, Any] = {"action": self.action} if self.by is not None: @@ -115,6 +186,10 @@ class ApprovalReceipt: result_summary: dict[str, Any] | None workflow_ref: str | None = None + @property + def response_provenance(self) -> dict[str, Any]: + return response_provenance_for(by=self.by, source=self.source) + # Neutral names for the general human/operator checkpoint substrate. Approval # remains a policy preset over this surface for now; these aliases let runtime, diff --git a/src/hermes_workflows/hermes_plugin_approvals.py b/src/hermes_workflows/hermes_plugin_approvals.py index d353674..6250f3b 100644 --- a/src/hermes_workflows/hermes_plugin_approvals.py +++ b/src/hermes_workflows/hermes_plugin_approvals.py @@ -7,9 +7,15 @@ from pathlib import Path from typing import Any, cast -from .approvals import ApprovalDecisionInput +from .approvals import ( + ApprovalDecisionInput, + is_revision_action_schema, + strip_client_controlled_provenance, + validate_revision_response, +) from .engine import WorkflowEngine from .receipts import redact_secrets +from .revision_validation import RevisionActionValidationError PLUGIN_NAME = "hermes-workflows-approvals" TOOLSET = "hermes_workflows_approvals" @@ -293,10 +299,31 @@ def _source_from_args(args: dict[str, Any]) -> dict[str, Any]: def _receipt_to_payload(receipt: Any, *, resume_requested: bool) -> dict[str, Any]: payload = _as_payload(receipt) + response_provenance = getattr(receipt, "response_provenance", None) + if isinstance(response_provenance, dict): + payload["response_provenance"] = response_provenance payload["resume_requested"] = resume_requested return payload +def _revision_schema_for_response(engine: WorkflowEngine, workflow_id: str, key: str) -> dict[str, Any] | None: + matching_step = next( + ( + step + for step in engine.list_operator_steps(status="waiting") + if step.get("workflow_id") == workflow_id and step.get("key") == key + ), + None, + ) + if not isinstance(matching_step, dict): + return None + descriptor = matching_step.get("schema_descriptor") + if not isinstance(descriptor, dict): + request = matching_step.get("request") + descriptor = request.get("schema_descriptor") if isinstance(request, dict) else None + return descriptor if isinstance(descriptor, dict) and is_revision_action_schema(descriptor) else None + + def _next_step_for_receipt(receipt_payload: dict[str, Any]) -> str | None: status = receipt_payload.get("status") if receipt_payload.get("resume_requested"): @@ -363,15 +390,29 @@ def _handle_workflow_review_respond(args: dict[str, Any], **kwargs: Any) -> str: payload = json.loads(payload) if not isinstance(payload, dict) or not payload: raise ValueError("payload must be a non-empty object") - receipt = WorkflowEngine(db_path).submit_operator_response( - workflow_id=str(args.get("workflow_id") or "").strip(), - key=str(args.get("key") or "").strip(), + workflow_id = str(args.get("workflow_id") or "").strip() + key = str(args.get("key") or "").strip() + payload = strip_client_controlled_provenance(payload) + engine = WorkflowEngine(db_path) + if _revision_schema_for_response(engine, workflow_id, key) is not None: + try: + payload, validated = validate_revision_response(payload) + except RevisionActionValidationError as exc: + return _tool_error(str(exc), validation=exc.to_dict()) + idempotency_key = f"revision:{workflow_id}:{key}:{validated.idempotency_key}" + else: + idempotency_key = ( + args.get("idempotency_key") + or args.get("message_url") + or args.get("message_id") + or args.get("event_id") + ) + receipt = engine.submit_operator_response( + workflow_id=workflow_id, + key=key, payload=payload, source=_source_from_args(args), - idempotency_key=args.get("idempotency_key") - or args.get("message_url") - or args.get("message_id") - or args.get("event_id"), + idempotency_key=idempotency_key, resume=resume, ) receipt_payload = _receipt_to_payload(receipt, resume_requested=resume) diff --git a/tests/test_approval_provenance.py b/tests/test_approval_provenance.py index 9539e8b..7d64455 100644 --- a/tests/test_approval_provenance.py +++ b/tests/test_approval_provenance.py @@ -111,6 +111,52 @@ def test_human_approval_accepts_human_source_and_returns_provenance(tmp_path): "channel": "discord", "message_url": "discord://thread/123/message/456", } + assert decision.to_dict() == { + "action": "approve", + "by": "skylar", + "source": { + "channel": "discord", + "message_url": "discord://thread/123/message/456", + }, + } + assert decision.response_provenance == { + "schema_version": 1, + "kind": "legacy_unverified", + "principal": None, + "display_label": "skylar", + "event": { + "channel": "discord", + "message_id": None, + "message_url": "discord://thread/123/message/456", + "event_id": None, + }, + } + + +def test_local_dashboard_decision_is_truthfully_unattributed(): + from hermes_workflows.approvals import ApprovalDecision + + decision = ApprovalDecision( + action="approve", + source={"channel": "local-dashboard", "event_id": "dashboard:click-1"}, + ) + + assert decision.to_dict() == { + "action": "approve", + "source": {"channel": "local-dashboard", "event_id": "dashboard:click-1"}, + } + assert decision.response_provenance == { + "schema_version": 1, + "kind": "unattributed_local_operator", + "principal": None, + "display_label": None, + "event": { + "channel": "local-dashboard", + "message_id": None, + "message_url": None, + "event_id": "dashboard:click-1", + }, + } def test_human_approval_rejects_missing_source(tmp_path): diff --git a/tests/test_dashboard_plugin.py b/tests/test_dashboard_plugin.py index 8bf5a28..dfcce72 100644 --- a/tests/test_dashboard_plugin.py +++ b/tests/test_dashboard_plugin.py @@ -740,9 +740,11 @@ def test_dashboard_approval_does_not_require_or_invent_actor_identity(tmp_path, assert receipt["success"] is True assert "by" not in receipt["receipt"] + assert receipt["receipt"]["response_provenance"]["kind"] == "unattributed_local_operator" + assert receipt["receipt"]["response_provenance"]["principal"] is None assert signal["payload"]["payload"] == {"action": "approve"} assert signal["payload"]["source"] == { - "channel": "hermes-dashboard", + "channel": "local-dashboard", "message_id": signal["payload"]["source"]["message_id"], } @@ -774,7 +776,7 @@ def test_dashboard_approval_strips_browser_actor_and_provenance(tmp_path, monkey assert receipt["success"] is True assert receipt["receipt"]["resume_requested"] is True assert signal["payload"]["source"] == { - "channel": "hermes-dashboard", + "channel": "local-dashboard", "message_id": signal["payload"]["source"]["message_id"], } assert "by" not in receipt["receipt"] @@ -810,8 +812,8 @@ def test_dashboard_approval_records_action_without_actor_identity(tmp_path, monk assert completed.status == "completed" status = WorkflowEngine(db).workflow_status("wf_named_human") signal = [event for event in status["events"] if event["type"] == "SignalReceived"][-1] - assert status["result"] == {"approved": True, "actor": None, "source_channel": "hermes-dashboard"} - assert signal["payload"]["source"]["channel"] == "hermes-dashboard" + assert status["result"] == {"approved": True, "actor": None, "source_channel": "local-dashboard"} + assert signal["payload"]["source"]["channel"] == "local-dashboard" assert signal["payload"]["payload"] == {"action": "approve"} @@ -844,8 +846,8 @@ def test_dashboard_approval_has_no_permission_or_actor_check(tmp_path, monkeypat assert completed.status == "completed" status = WorkflowEngine(db).workflow_status("wf_named_human_mismatch") signal = [event for event in status["events"] if event["type"] == "SignalReceived"][-1] - assert status["result"] == {"approved": True, "actor": None, "source_channel": "hermes-dashboard"} - assert signal["payload"]["source"]["channel"] == "hermes-dashboard" + assert status["result"] == {"approved": True, "actor": None, "source_channel": "local-dashboard"} + assert signal["payload"]["source"]["channel"] == "local-dashboard" assert signal["payload"]["payload"] == {"action": "approve"} @@ -885,11 +887,81 @@ def test_dashboard_review_response_does_not_require_or_invent_approver_identity( signal = [event for event in WorkflowEngine(db).events("wf_dashboard_ask_response") if event["type"] == "SignalReceived"][-1] assert signal["payload"]["payload"] == {"action": "approve"} assert signal["payload"]["source"] == { - "channel": "hermes-dashboard", + "channel": "local-dashboard", "message_id": signal["payload"]["source"]["message_id"], } +def test_dashboard_review_response_rejects_empty_request_changes_with_structured_error(tmp_path, monkeypatch): + db = tmp_path / "workflow.sqlite" + WorkflowEngine(db).run_until_idle( + dashboard_review_decision_workflow, + {}, + workflow_id="wf_dashboard_empty_revision", + workflow_ref="tests.test_dashboard_plugin:dashboard_review_decision_workflow", + ) + configure_test_dbs(monkeypatch, tmp_path, {"runtime-smoke": str(db)}) + api = load_dashboard_api() + + with pytest.raises(Exception) as excinfo: + run( + api.respond_review_request( + { + "db": "runtime-smoke", + "workflow_id": "wf_dashboard_empty_revision", + "key": "review_dashboard_decision", + "payload": {"action": "request_changes", "feedback": "\u2003"}, + } + ) + ) + + assert getattr(excinfo.value, "status_code", None) == 400 + detail = getattr(excinfo.value, "detail", None) + assert isinstance(detail, dict) + assert detail["code"] == "revision_action_invalid" + assert detail["message"] == "request_changes requires nonblank feedback or valid edited_output" + assert [error["field"] for error in detail["field_errors"]] == ["feedback", "edited_output"] + assert WorkflowEngine(db).workflow_status("wf_dashboard_empty_revision")["status"] == "waiting" + assert not [event for event in WorkflowEngine(db).events("wf_dashboard_empty_revision") if event["type"] == "SignalReceived"] + + +def test_dashboard_review_response_strips_all_client_identity_and_provenance_fields(tmp_path, monkeypatch): + db = tmp_path / "workflow.sqlite" + WorkflowEngine(db).run_until_idle( + dashboard_review_decision_workflow, + {}, + workflow_id="wf_dashboard_spoof_revision", + workflow_ref="tests.test_dashboard_plugin:dashboard_review_decision_workflow", + ) + configure_test_dbs(monkeypatch, tmp_path, {"runtime-smoke": str(db)}) + api = load_dashboard_api() + + receipt = run( + api.respond_review_request( + { + "db": "runtime-smoke", + "workflow_id": "wf_dashboard_spoof_revision", + "key": "review_dashboard_decision", + "payload": { + "action": "request_changes", + "feedback": " Tighten it. ", + "by": "skylar", + "actor": "skylar", + "principal": {"subject": "skylar"}, + "source": {"id": "skylar"}, + "response_provenance": {"kind": "authenticated_principal", "principal": {"subject": "skylar"}}, + }, + } + ) + ) + + assert receipt["receipt"]["response_provenance"]["kind"] == "unattributed_local_operator" + assert receipt["receipt"]["response_provenance"]["principal"] is None + signal = [event for event in WorkflowEngine(db).events("wf_dashboard_spoof_revision") if event["type"] == "SignalReceived"][-1] + assert signal["payload"]["payload"] == {"action": "request_changes", "feedback": "Tighten it."} + assert signal["payload"]["source"]["channel"] == "local-dashboard" + + def test_dashboard_review_response_idempotency_key_replay_does_not_duplicate_continuation(tmp_path, monkeypatch): db = tmp_path / "workflow.sqlite" WorkflowEngine(db).run_until_idle( @@ -1200,6 +1272,7 @@ def test_dashboard_frontend_renders_structured_forms_instead_of_raw_trip_json_bo assert "hwf-structured-field" in index_js assert "hwf-structured-form" in style_css assert "hwf-structured-field-type" in style_css + assert "request_changes requires nonblank feedback or valid edited_output" in index_js diff --git a/tests/test_hermes_plugin_approvals.py b/tests/test_hermes_plugin_approvals.py index 50895e0..30440c0 100644 --- a/tests/test_hermes_plugin_approvals.py +++ b/tests/test_hermes_plugin_approvals.py @@ -11,7 +11,7 @@ import pytest -from hermes_workflows import WorkflowEngine, approve, step, workflow +from hermes_workflows import WorkflowEngine, approve, ask, step, workflow from hermes_workflows.engine import JsonCodec @@ -34,6 +34,24 @@ async def plugin_approval_workflow(inputs): return await plugin_followup_step(decision) +@dataclass +class PluginReviewDecision: + action: str + feedback: str | None = None + edited_output: str | None = None + + +@workflow +async def plugin_review_workflow(inputs): + return await ask( + "Review the plugin draft?", + key="review_plugin_draft", + input={"draft": "test"}, + choice=["approve", "request_changes"], + returns=PluginReviewDecision, + ) + + class FakePluginContext: def __init__(self): self.tools: dict[str, dict[str, Any]] = {} @@ -61,6 +79,17 @@ def create_pending_approval(db: Path) -> WorkflowEngine: return engine +def create_pending_review(db: Path) -> WorkflowEngine: + engine = WorkflowEngine(db) + engine.run_until_idle( + plugin_review_workflow, + {}, + workflow_id="wf_plugin_review", + workflow_ref="tests.test_hermes_plugin_approvals:plugin_review_workflow", + ) + return engine + + def parse_tool_result(raw: str) -> dict[str, Any]: data = json.loads(raw) assert data["success"] is True @@ -213,6 +242,9 @@ def test_workflow_approval_decide_defaults_to_resume_true(tmp_path): ) assert result["receipt"]["action"] == "approve" + assert result["receipt"]["response_provenance"]["kind"] == "legacy_unverified" + assert result["receipt"]["response_provenance"]["principal"] is None + assert result["receipt"]["response_provenance"]["display_label"] == "operator" assert result["receipt"]["resume_requested"] is True assert result["receipt"]["status"] == "running" completed = WorkflowEngine(db).drain("wf_plugin") @@ -220,6 +252,86 @@ def test_workflow_approval_decide_defaults_to_resume_true(tmp_path): assert completed.result["followup_ran"] is True +def test_workflow_review_respond_rejects_empty_request_changes_with_contract_error(tmp_path): + from hermes_workflows.hermes_plugin_approvals import register + + db = tmp_path / "workflow.sqlite" + create_pending_review(db) + plugin_context = FakePluginContext() + register(plugin_context) + + result = json.loads( + plugin_context.tools["workflow_review_respond"]["handler"]( + { + "db": str(db), + "workflow_id": "wf_plugin_review", + "key": "review_plugin_draft", + "payload": {"action": "request_changes", "feedback": "\u2003"}, + "by": "skylar", + "channel": "hermes-plugin", + "message_id": "tool-review-1", + } + ) + ) + + assert result["success"] is False + assert result["validation"] == { + "code": "revision_action_invalid", + "message": "request_changes requires nonblank feedback or valid edited_output", + "field_errors": [ + { + "field": "feedback", + "code": "actionable_required", + "message": "request_changes requires nonblank feedback or valid edited_output", + }, + { + "field": "edited_output", + "code": "actionable_required", + "message": "request_changes requires nonblank feedback or valid edited_output", + }, + ], + } + assert WorkflowEngine(db).workflow_status("wf_plugin_review")["status"] == "waiting" + assert not [event for event in WorkflowEngine(db).events("wf_plugin_review") if event["type"] == "SignalReceived"] + + +def test_workflow_review_respond_strips_client_identity_and_records_normalized_revision(tmp_path): + from hermes_workflows.hermes_plugin_approvals import register + + db = tmp_path / "workflow.sqlite" + create_pending_review(db) + plugin_context = FakePluginContext() + register(plugin_context) + + result = parse_tool_result( + plugin_context.tools["workflow_review_respond"]["handler"]( + { + "db": str(db), + "workflow_id": "wf_plugin_review", + "key": "review_plugin_draft", + "payload": { + "action": " request_changes ", + "feedback": " Make it concrete. ", + "by": "skylar", + "actor": "skylar", + "source": {"id": "skylar"}, + "response_provenance": {"kind": "authenticated_principal", "principal": {"subject": "skylar"}}, + }, + "by": "skylar", + "channel": "hermes-plugin", + "message_id": "tool-review-2", + "resume": False, + } + ) + ) + + assert result["receipt"]["response_provenance"]["kind"] == "legacy_unverified" + assert result["receipt"]["response_provenance"]["principal"] is None + signal = [event for event in WorkflowEngine(db).events("wf_plugin_review") if event["type"] == "SignalReceived"][-1] + assert signal["payload"]["payload"] == {"action": "request_changes", "feedback": "Make it concrete."} + assert signal["idempotency_key"].startswith("revision:wf_plugin_review:review_plugin_draft:revision-action:v1:") + + def test_workflow_approval_decide_can_resume_when_explicitly_requested(tmp_path): from hermes_workflows.hermes_plugin_approvals import register From 0afe40754c64b14239f3c29448e1d3d355494f05 Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Tue, 14 Jul 2026 11:18:24 -0700 Subject: [PATCH 2/3] INT-O1: preserve normalized revision replay --- .../hermes_plugin_approvals.py | 15 +++---- tests/test_dashboard_plugin.py | 44 +++++++++++++++++++ tests/test_hermes_plugin_approvals.py | 40 +++++++++++++++++ 3 files changed, 90 insertions(+), 9 deletions(-) diff --git a/src/hermes_workflows/hermes_plugin_approvals.py b/src/hermes_workflows/hermes_plugin_approvals.py index 6250f3b..5ae2ada 100644 --- a/src/hermes_workflows/hermes_plugin_approvals.py +++ b/src/hermes_workflows/hermes_plugin_approvals.py @@ -307,20 +307,17 @@ def _receipt_to_payload(receipt: Any, *, resume_requested: bool) -> dict[str, An def _revision_schema_for_response(engine: WorkflowEngine, workflow_id: str, key: str) -> dict[str, Any] | None: - matching_step = next( + matching_request = next( ( - step - for step in engine.list_operator_steps(status="waiting") - if step.get("workflow_id") == workflow_id and step.get("key") == key + event.get("payload") + for event in reversed(engine.events(workflow_id)) + if event.get("type") == "ApprovalRequested" and event.get("key") == f"approval:{key}" ), None, ) - if not isinstance(matching_step, dict): + if not isinstance(matching_request, dict): return None - descriptor = matching_step.get("schema_descriptor") - if not isinstance(descriptor, dict): - request = matching_step.get("request") - descriptor = request.get("schema_descriptor") if isinstance(request, dict) else None + descriptor = matching_request.get("schema_descriptor") return descriptor if isinstance(descriptor, dict) and is_revision_action_schema(descriptor) else None diff --git a/tests/test_dashboard_plugin.py b/tests/test_dashboard_plugin.py index dfcce72..87db2d9 100644 --- a/tests/test_dashboard_plugin.py +++ b/tests/test_dashboard_plugin.py @@ -962,6 +962,50 @@ def test_dashboard_review_response_strips_all_client_identity_and_provenance_fie assert signal["payload"]["source"]["channel"] == "local-dashboard" +def test_dashboard_review_response_replays_normalized_revision_after_step_completion(tmp_path, monkeypatch): + db = tmp_path / "workflow.sqlite" + WorkflowEngine(db).run_until_idle( + dashboard_review_decision_workflow, + {}, + workflow_id="wf_dashboard_normalized_revision_replay", + workflow_ref="tests.test_dashboard_plugin:dashboard_review_decision_workflow", + ) + configure_test_dbs(monkeypatch, tmp_path, {"runtime-smoke": str(db)}) + api = load_dashboard_api() + body = { + "db": "runtime-smoke", + "workflow_id": "wf_dashboard_normalized_revision_replay", + "key": "review_dashboard_decision", + "payload": {"action": " request_changes ", "feedback": " Tighten the opening. "}, + "idempotency_key": "dashboard-normalized-replay-1", + } + + first = run(api.respond_review_request(body)) + replay_body = dict(body) + replay_body["payload"] = {"action": "request_changes", "feedback": "Tighten the opening."} + second = run(api.respond_review_request(replay_body)) + + assert first["success"] is True + assert second["success"] is True + assert dashboard_operator_response_counts(db, "wf_dashboard_normalized_revision_replay", "review_dashboard_decision") == { + "signals": 1, + "steps": 1, + "commands": 1, + "command_status": "pending", + } + + conflicting_body = dict(body) + conflicting_body["payload"] = {"action": "request_changes", "feedback": "Use a different structure."} + with pytest.raises(Exception, match="already has a recorded decision/response"): + run(api.respond_review_request(conflicting_body)) + assert dashboard_operator_response_counts(db, "wf_dashboard_normalized_revision_replay", "review_dashboard_decision") == { + "signals": 1, + "steps": 1, + "commands": 1, + "command_status": "pending", + } + + def test_dashboard_review_response_idempotency_key_replay_does_not_duplicate_continuation(tmp_path, monkeypatch): db = tmp_path / "workflow.sqlite" WorkflowEngine(db).run_until_idle( diff --git a/tests/test_hermes_plugin_approvals.py b/tests/test_hermes_plugin_approvals.py index 30440c0..b3e4385 100644 --- a/tests/test_hermes_plugin_approvals.py +++ b/tests/test_hermes_plugin_approvals.py @@ -332,6 +332,46 @@ def test_workflow_review_respond_strips_client_identity_and_records_normalized_r assert signal["idempotency_key"].startswith("revision:wf_plugin_review:review_plugin_draft:revision-action:v1:") +def test_workflow_review_respond_replays_normalized_revision_after_step_completion(tmp_path): + from hermes_workflows.hermes_plugin_approvals import register + + db = tmp_path / "workflow.sqlite" + create_pending_review(db) + plugin_context = FakePluginContext() + register(plugin_context) + handler = plugin_context.tools["workflow_review_respond"]["handler"] + body = { + "db": str(db), + "workflow_id": "wf_plugin_review", + "key": "review_plugin_draft", + "payload": {"action": " request_changes ", "feedback": " Make it concrete. "}, + "by": "operator", + "channel": "hermes-plugin", + "message_id": "tool-normalized-replay-1", + "resume": False, + } + + first = parse_tool_result(handler(body)) + replay_body = dict(body) + replay_body["payload"] = {"action": "request_changes", "feedback": "Make it concrete."} + second = parse_tool_result(handler(replay_body)) + + assert first["success"] is True + assert second["success"] is True + events = WorkflowEngine(db).events("wf_plugin_review") + assert len([event for event in events if event["type"] == "SignalReceived" and event["key"] == "signal:operator.response:review_plugin_draft"]) == 1 + assert len([event for event in events if event["type"] == "StepCompleted" and event["key"] == "review_plugin_draft"]) == 1 + + conflicting_body = dict(body) + conflicting_body["payload"] = {"action": "request_changes", "feedback": "Use a different structure."} + conflicting = json.loads(handler(conflicting_body)) + assert conflicting["success"] is False + assert "already has a recorded decision/response" in conflicting["error"] + events = WorkflowEngine(db).events("wf_plugin_review") + assert len([event for event in events if event["type"] == "SignalReceived" and event["key"] == "signal:operator.response:review_plugin_draft"]) == 1 + assert len([event for event in events if event["type"] == "StepCompleted" and event["key"] == "review_plugin_draft"]) == 1 + + def test_workflow_approval_decide_can_resume_when_explicitly_requested(tmp_path): from hermes_workflows.hermes_plugin_approvals import register From a2005f55abec05a3f18ed87522fbe43620f770aa Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Tue, 14 Jul 2026 11:54:40 -0700 Subject: [PATCH 3/3] INT-O1: replay revisions across transport IDs --- .../dashboard/plugin_api.py | 17 +++++++-- .../hermes_plugin_approvals.py | 37 ++++++++++++++++++- tests/test_dashboard_plugin.py | 13 +++++++ tests/test_hermes_plugin_approvals.py | 11 +++++- 4 files changed, 72 insertions(+), 6 deletions(-) diff --git a/plugins/hermes-workflows-approvals/dashboard/plugin_api.py b/plugins/hermes-workflows-approvals/dashboard/plugin_api.py index 309ca27..b567cb7 100644 --- a/plugins/hermes-workflows-approvals/dashboard/plugin_api.py +++ b/plugins/hermes-workflows-approvals/dashboard/plugin_api.py @@ -21,6 +21,7 @@ _redact, _receipt_to_payload, _revision_schema_for_response, + _source_for_normalized_revision_replay, approval_view_to_dict, ) from hermes_workflows.revision_validation import RevisionActionValidationError @@ -2254,17 +2255,25 @@ def record_and_resume() -> tuple[Any, dict[str, Any]]: engine = WorkflowEngine(db_path) normalized_payload = _normalize_review_payload_for_dashboard_request(engine, workflow_id, key, payload) effective_idempotency_key = message_id + source = { + "channel": "local-dashboard", + "message_id": message_id, + } if _revision_schema_for_response(engine, workflow_id, key) is not None: normalized_payload, validated = validate_revision_response(normalized_payload) effective_idempotency_key = f"revision:{workflow_id}:{key}:{validated.idempotency_key}" + source = _source_for_normalized_revision_replay( + engine, + workflow_id, + key, + effective_idempotency_key, + source, + ) receipt = engine.submit_operator_response( workflow_id=workflow_id, key=key, payload=normalized_payload, - source={ - "channel": "local-dashboard", - "message_id": message_id, - }, + source=source, idempotency_key=effective_idempotency_key, resume=True, ) diff --git a/src/hermes_workflows/hermes_plugin_approvals.py b/src/hermes_workflows/hermes_plugin_approvals.py index 5ae2ada..87ce3b4 100644 --- a/src/hermes_workflows/hermes_plugin_approvals.py +++ b/src/hermes_workflows/hermes_plugin_approvals.py @@ -321,6 +321,33 @@ def _revision_schema_for_response(engine: WorkflowEngine, workflow_id: str, key: return descriptor if isinstance(descriptor, dict) and is_revision_action_schema(descriptor) else None +def _source_for_normalized_revision_replay( + engine: WorkflowEngine, + workflow_id: str, + key: str, + idempotency_key: str, + source: dict[str, Any], +) -> dict[str, Any]: + """Keep the first durable transport provenance for an exact revision replay.""" + + signal_key = f"signal:operator.response:{key}" + matching_signal = next( + ( + event + for event in reversed(engine.events(workflow_id)) + if event.get("type") == "SignalReceived" + and event.get("key") == signal_key + and event.get("idempotency_key") == idempotency_key + ), + None, + ) + if not isinstance(matching_signal, dict): + return source + event_payload = matching_signal.get("payload") + persisted_source = event_payload.get("source") if isinstance(event_payload, dict) else None + return dict(persisted_source) if isinstance(persisted_source, dict) else source + + def _next_step_for_receipt(receipt_payload: dict[str, Any]) -> str | None: status = receipt_payload.get("status") if receipt_payload.get("resume_requested"): @@ -391,12 +418,20 @@ def _handle_workflow_review_respond(args: dict[str, Any], **kwargs: Any) -> str: key = str(args.get("key") or "").strip() payload = strip_client_controlled_provenance(payload) engine = WorkflowEngine(db_path) + source = _source_from_args(args) if _revision_schema_for_response(engine, workflow_id, key) is not None: try: payload, validated = validate_revision_response(payload) except RevisionActionValidationError as exc: return _tool_error(str(exc), validation=exc.to_dict()) idempotency_key = f"revision:{workflow_id}:{key}:{validated.idempotency_key}" + source = _source_for_normalized_revision_replay( + engine, + workflow_id, + key, + idempotency_key, + source, + ) else: idempotency_key = ( args.get("idempotency_key") @@ -408,7 +443,7 @@ def _handle_workflow_review_respond(args: dict[str, Any], **kwargs: Any) -> str: workflow_id=workflow_id, key=key, payload=payload, - source=_source_from_args(args), + source=source, idempotency_key=idempotency_key, resume=resume, ) diff --git a/tests/test_dashboard_plugin.py b/tests/test_dashboard_plugin.py index 87db2d9..174d6ca 100644 --- a/tests/test_dashboard_plugin.py +++ b/tests/test_dashboard_plugin.py @@ -983,19 +983,32 @@ def test_dashboard_review_response_replays_normalized_revision_after_step_comple first = run(api.respond_review_request(body)) replay_body = dict(body) replay_body["payload"] = {"action": "request_changes", "feedback": "Tighten the opening."} + replay_body["idempotency_key"] = "dashboard-normalized-replay-2" second = run(api.respond_review_request(replay_body)) assert first["success"] is True assert second["success"] is True + assert second["receipt"]["response_provenance"]["kind"] == "unattributed_local_operator" + assert second["receipt"]["response_provenance"]["principal"] is None assert dashboard_operator_response_counts(db, "wf_dashboard_normalized_revision_replay", "review_dashboard_decision") == { "signals": 1, "steps": 1, "commands": 1, "command_status": "pending", } + signals = [ + event + for event in WorkflowEngine(db).events("wf_dashboard_normalized_revision_replay") + if event["type"] == "SignalReceived" and event["key"] == "signal:operator.response:review_dashboard_decision" + ] + assert signals[0]["payload"]["source"] == { + "channel": "local-dashboard", + "message_id": "dashboard:dashboard-normalized-replay-1", + } conflicting_body = dict(body) conflicting_body["payload"] = {"action": "request_changes", "feedback": "Use a different structure."} + conflicting_body["idempotency_key"] = "dashboard-normalized-replay-3" with pytest.raises(Exception, match="already has a recorded decision/response"): run(api.respond_review_request(conflicting_body)) assert dashboard_operator_response_counts(db, "wf_dashboard_normalized_revision_replay", "review_dashboard_decision") == { diff --git a/tests/test_hermes_plugin_approvals.py b/tests/test_hermes_plugin_approvals.py index b3e4385..275c225 100644 --- a/tests/test_hermes_plugin_approvals.py +++ b/tests/test_hermes_plugin_approvals.py @@ -354,16 +354,25 @@ def test_workflow_review_respond_replays_normalized_revision_after_step_completi first = parse_tool_result(handler(body)) replay_body = dict(body) replay_body["payload"] = {"action": "request_changes", "feedback": "Make it concrete."} + replay_body["message_id"] = "tool-normalized-replay-2" second = parse_tool_result(handler(replay_body)) assert first["success"] is True assert second["success"] is True + assert second["receipt"]["response_provenance"]["kind"] == "legacy_unverified" + assert second["receipt"]["response_provenance"]["principal"] is None events = WorkflowEngine(db).events("wf_plugin_review") - assert len([event for event in events if event["type"] == "SignalReceived" and event["key"] == "signal:operator.response:review_plugin_draft"]) == 1 + signals = [event for event in events if event["type"] == "SignalReceived" and event["key"] == "signal:operator.response:review_plugin_draft"] + assert len(signals) == 1 + assert signals[0]["payload"]["source"] == { + "channel": "hermes-plugin", + "message_id": "tool-normalized-replay-1", + } assert len([event for event in events if event["type"] == "StepCompleted" and event["key"] == "review_plugin_draft"]) == 1 conflicting_body = dict(body) conflicting_body["payload"] = {"action": "request_changes", "feedback": "Use a different structure."} + conflicting_body["message_id"] = "tool-normalized-replay-3" conflicting = json.loads(handler(conflicting_body)) assert conflicting["success"] is False assert "already has a recorded decision/response" in conflicting["error"]