diff --git a/src/titmas_action_gate/agents.py b/src/titmas_action_gate/agents.py index 9416f07..d6792c1 100644 --- a/src/titmas_action_gate/agents.py +++ b/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] diff --git a/src/titmas_action_gate/github_sandbox.py b/src/titmas_action_gate/github_sandbox.py index 7dc83f7..b1ea400 100644 --- a/src/titmas_action_gate/github_sandbox.py +++ b/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"], diff --git a/src/titmas_action_gate/service.py b/src/titmas_action_gate/service.py index 90d7164..df6bc08 100644 --- a/src/titmas_action_gate/service.py +++ b/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) diff --git a/src/titmas_action_gate/workflow.py b/src/titmas_action_gate/workflow.py index 5902519..08067cb 100644 --- a/src/titmas_action_gate/workflow.py +++ b/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,9 +98,25 @@ 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"], @@ -100,7 +124,15 @@ def run(self, *, repository: str, base_time: datetime | None = None) -> dict[str 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"], diff --git a/tests/test_security_argument_injection.py b/tests/test_security_argument_injection.py index 3990256..4d8bb42 100644 --- a/tests/test_security_argument_injection.py +++ b/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") diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 61c356d..461a3bf 100644 --- a/tests/test_workflow.py +++ b/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)