🧹 Refactor HandoffLog.add to take Handoff object

src/titmas_action_gate/agents.py

@@ -25,16 +25,8 @@ class Handoff:
 class HandoffLog:
     entries: list[Handoff] = field(default_factory=list)
 
-    def add(self, sender: str, recipient: str, request_id: str, responsibility: str, payload: dict[str, Any]) -> None:
-        self.entries.append(
-            Handoff(
-                sender=sender,
-                recipient=recipient,
-                request_id=request_id,
-                responsibility=responsibility,
-                payload_sha256=sha256_json(payload),
-            )
-        )
+    def add(self, handoff: Handoff) -> None:
+        self.entries.append(handoff)
 
     def as_dicts(self) -> list[dict[str, str]]:
         return [entry.__dict__.copy() for entry in self.entries]

src/titmas_action_gate/github_sandbox.py

@@ -7,7 +7,8 @@
 from pathlib import Path
 from typing import Any
 
-from .agents import EvidenceVerifier, GitHubOperator, HandoffLog, RequestAnalyst, WorkflowLead
+from .agents import EvidenceVerifier, GitHubOperator, Handoff, HandoffLog, RequestAnalyst, WorkflowLead
+from .canonical import sha256_json
 from .provider import GhCliProvider
 from .service import ActionGateService
 from .workflow import AGENTTEAMS_COMMIT, AGENTTEAMS_RELEASE
@@ -83,14 +84,30 @@ def run(self, *, branch: str, base: str = "main", title: str = "TITMAS Agent Act
             uncertainty=[],
             created_at=observed,
         )
-        self.handoffs.add("workflow-lead", "request-analyst", push_request["request_id"], "NORMALIZE_BRANCH_PUSH", push_request)
+        self.handoffs.add(
+            Handoff(
+                sender="workflow-lead",
+                recipient="request-analyst",
+                request_id=push_request["request_id"],
+                responsibility="NORMALIZE_BRANCH_PUSH",
+                payload_sha256=sha256_json(push_request),
+            )
+        )
         push_gate = self._verify_and_decide(
             push_request,
             phase="branch-push-pre-execution",
             evidence_output={"commit": commit, "worktree": "LOCAL_PATH_REDACTED_IN_PUBLIC_REPORT"},
             observed=observed + timedelta(seconds=1),
         )
-        self.handoffs.add("workflow-lead", "github-operator", push_request["request_id"], "EXECUTE_EXACT_BRANCH_PUSH", push_gate["decision"])
+        self.handoffs.add(
+            Handoff(
+                sender="workflow-lead",
+                recipient="github-operator",
+                request_id=push_request["request_id"],
+                responsibility="EXECUTE_EXACT_BRANCH_PUSH",
+                payload_sha256=sha256_json(push_gate["decision"]),
+            )
+        )
         push_receipt = self.operator.execute(
             self.service,
             push_request["request_id"],
@@ -109,14 +126,30 @@ def run(self, *, branch: str, base: str = "main", title: str = "TITMAS Agent Act
             uncertainty=[],
             created_at=observed + timedelta(seconds=4),
         )
-        self.handoffs.add("github-operator", "release-steward", pr_request["request_id"], "PREPARE_PR_REQUEST", push_receipt)
+        self.handoffs.add(
+            Handoff(
+                sender="github-operator",
+                recipient="release-steward",
+                request_id=pr_request["request_id"],
+                responsibility="PREPARE_PR_REQUEST",
+                payload_sha256=sha256_json(push_receipt),
+            )
+        )
         pr_gate = self._verify_and_decide(
             pr_request,
             phase="pull-request-pre-execution",
             evidence_output=push_receipt,
             observed=observed + timedelta(seconds=5),
         )
-        self.handoffs.add("workflow-lead", "github-operator", pr_request["request_id"], "EXECUTE_EXACT_PR_CREATE", pr_gate["decision"])
+        self.handoffs.add(
+            Handoff(
+                sender="workflow-lead",
+                recipient="github-operator",
+                request_id=pr_request["request_id"],
+                responsibility="EXECUTE_EXACT_PR_CREATE",
+                payload_sha256=sha256_json(pr_gate["decision"]),
+            )
+        )
         pr_receipt = self.operator.execute(
             self.service,
             pr_request["request_id"],

src/titmas_action_gate/service.py

@@ -5,6 +5,7 @@
 import hashlib
 import hmac
 import json
+from dataclasses import dataclass
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any
@@ -20,7 +21,6 @@
 from .provider import GitHubProvider
 from .signing import HmacRecordSigner
 from .store import AppendOnlyStore
-from dataclasses import dataclass
 
 
 @dataclass(frozen=True)

src/titmas_action_gate/workflow.py

@@ -9,8 +9,8 @@
 
 import yaml
 
-from .agents import EvidenceVerifier, GitHubOperator, HandoffLog, ReleaseSteward, RequestAnalyst, WorkflowLead
-from .canonical import format_datetime
+from .agents import EvidenceVerifier, GitHubOperator, Handoff, HandoffLog, ReleaseSteward, RequestAnalyst, WorkflowLead
+from .canonical import format_datetime, sha256_json
 from .provider import InMemoryGitHubProvider
 from .service import ActionGateService
 
@@ -71,7 +71,15 @@ def run(self, *, repository: str, base_time: datetime | None = None) -> dict[str
             uncertainty=["Provider execution is sandboxed; no external GitHub write is claimed."],
             created_at=observed,
         )
-        self.handoffs.add("workflow-lead", "request-analyst", request["request_id"], "NORMALIZE_REQUEST", request)
+        self.handoffs.add(
+            Handoff(
+                sender="workflow-lead",
+                recipient="request-analyst",
+                request_id=request["request_id"],
+                responsibility="NORMALIZE_REQUEST",
+                payload_sha256=sha256_json(request),
+            )
+        )
         self.service.submit_action_request(request, caller_token=self.caller_token)
 
         profile = self.service.generate_evidence_profile(
@@ -90,17 +98,41 @@ def run(self, *, repository: str, base_time: datetime | None = None) -> dict[str
             request["evidence_requirements"],
             caller_token=self.caller_token,
         )
-        self.handoffs.add("workflow-lead", "evidence-verifier", request["request_id"], "VERIFY_PRE_EXECUTION_EVIDENCE", profile)
+        self.handoffs.add(
+            Handoff(
+                sender="workflow-lead",
+                recipient="evidence-verifier",
+                request_id=request["request_id"],
+                responsibility="VERIFY_PRE_EXECUTION_EVIDENCE",
+                payload_sha256=sha256_json(profile),
+            )
+        )
         evidence_result = self.verifier.verify(self.service, request["request_id"], caller_token=self.caller_token)
-        self.handoffs.add("evidence-verifier", "workflow-lead", request["request_id"], "RETURN_VERIFIER_RECEIPT", evidence_result)
+        self.handoffs.add(
+            Handoff(
+                sender="evidence-verifier",
+                recipient="workflow-lead",
+                request_id=request["request_id"],
+                responsibility="RETURN_VERIFIER_RECEIPT",
+                payload_sha256=sha256_json(evidence_result),
+            )
+        )
         initial_envelope = self.lead.decide(
             self.service,
             request["request_id"],
             caller_token=self.caller_token,
             decided_at=observed + timedelta(seconds=2),
         )
         initial_decision = initial_envelope["payload"]
-        self.handoffs.add("workflow-lead", "github-operator", request["request_id"], "EXECUTE_EXACT_ALLOW", initial_decision)
+        self.handoffs.add(
+            Handoff(
+                sender="workflow-lead",
+                recipient="github-operator",
+                request_id=request["request_id"],
+                responsibility="EXECUTE_EXACT_ALLOW",
+                payload_sha256=sha256_json(initial_decision),
+            )
+        )
         execution_receipt = self.operator.execute(
             self.service,
             request["request_id"],
@@ -117,7 +149,15 @@ def run(self, *, repository: str, base_time: datetime | None = None) -> dict[str
             pull_number=pull_number,
             created_at=observed + timedelta(seconds=4),
         )
-        self.handoffs.add("github-operator", "release-steward", release_request["request_id"], "ASSEMBLE_POST_EXECUTION_EVIDENCE", execution_receipt)
+        self.handoffs.add(
+            Handoff(
+                sender="github-operator",
+                recipient="release-steward",
+                request_id=release_request["request_id"],
+                responsibility="ASSEMBLE_POST_EXECUTION_EVIDENCE",
+                payload_sha256=sha256_json(execution_receipt),
+            )
+        )
         self.service.submit_action_request(release_request, caller_token=self.caller_token)
         post_profile = self.service.generate_evidence_profile(
             release_request["request_id"],
@@ -142,7 +182,15 @@ def run(self, *, repository: str, base_time: datetime | None = None) -> dict[str
             caller_token=self.caller_token,
             decided_at=observed + timedelta(seconds=6),
         )["payload"]
-        self.handoffs.add("workflow-lead", "titmas-action-gate-reviewer", release_request["request_id"], "REQUEST_SCOPED_HUMAN_APPROVAL", before_approval)
+        self.handoffs.add(
+            Handoff(
+                sender="workflow-lead",
+                recipient="titmas-action-gate-reviewer",
+                request_id=release_request["request_id"],
+                responsibility="REQUEST_SCOPED_HUMAN_APPROVAL",
+                payload_sha256=sha256_json(before_approval),
+            )
+        )
         approval = self.service.record_human_approval(
             release_request["request_id"],
             subject="human:demo-reviewer",
@@ -151,7 +199,15 @@ def run(self, *, repository: str, base_time: datetime | None = None) -> dict[str
             approver_token=self.approver_token,
             decided_at=observed + timedelta(seconds=7),
         )
-        self.handoffs.add("titmas-action-gate-reviewer", "workflow-lead", release_request["request_id"], "RETURN_SCOPED_APPROVAL", approval)
+        self.handoffs.add(
+            Handoff(
+                sender="titmas-action-gate-reviewer",
+                recipient="workflow-lead",
+                request_id=release_request["request_id"],
+                responsibility="RETURN_SCOPED_APPROVAL",
+                payload_sha256=sha256_json(approval),
+            )
+        )
         after_approval = self.lead.decide(
             self.service,
             release_request["request_id"],

tests/test_security_argument_injection.py

@@ -1,9 +1,8 @@
 import unittest
-import subprocess
-from unittest.mock import patch, MagicMock
+from unittest.mock import MagicMock, patch
 
 from titmas_action_gate.provider import GhCliProvider
-from titmas_action_gate.errors import ActionGateError
+
 
 class ProviderSecurityTests(unittest.TestCase):
     @patch("subprocess.run")

tests/test_workflow.py

@@ -1,8 +1,8 @@
+import json
 import tempfile
 import unittest
 from pathlib import Path
 
-import json
 from titmas_action_gate.workflow import validate_agentteams_template, write_demo_report
 
 
@@ -83,7 +83,7 @@ def test_write_demo_report(self):
         self.assertEqual(result_path, output_file)
         self.assertTrue(output_file.exists())
 
-        with open(output_file, "r", encoding="utf-8") as f:
+        with open(output_file, encoding="utf-8") as f:
             content = json.load(f)
 
         self.assertEqual(content, report_data)
@@ -95,7 +95,7 @@ def test_write_demo_report_creates_directories(self):
         write_demo_report(report_data, output_file)
 
         self.assertTrue(output_file.exists())
-        with open(output_file, "r", encoding="utf-8") as f:
+        with open(output_file, encoding="utf-8") as f:
             content = json.load(f)
 
         self.assertEqual(content, report_data)