feat: introduce Task Contract and Plan–Act–Evaluate artifacts

app/agent/graph/assessment_graph.py

@@ -10,6 +10,7 @@
 
 from app.core.config import settings
 from app.core.db import engine
+from app.models.agent_execution import AgentTaskContract
 from app.models.assessment import AssessmentReport
 from app.models.governance import ControlEvidenceItem, ControlInstance
 from app.models.parser import ParsedDocument
@@ -141,6 +142,17 @@ async def _build_context(state: AssessmentGraphState) -> AssessmentGraphState:
     }
 
 
+async def _plan(state: AssessmentGraphState) -> AssessmentGraphState:
+    from app.agent.task_contract import build_plan_artifact
+
+    return {
+        "plan_artifact": build_plan_artifact(
+            state["task_contract"],
+            skill_id=state.get("skill_id"),
+        )
+    }
+
+
 async def _gather_context(state: AssessmentGraphState) -> AssessmentGraphState:
     from app.agent import orchestrator as legacy
 
@@ -233,6 +245,23 @@ async def _verify_threat_evidence(
     return {"report": report}
 
 
+async def _evaluate(state: AssessmentGraphState) -> AssessmentGraphState:
+    from app.agent.task_contract import evaluate_assessment_report
+
+    evaluation = evaluate_assessment_report(
+        state["report"],
+        state["task_contract"],
+    )
+    report = state["report"].model_copy(
+        update={
+            "task_contract": state["task_contract"],
+            "plan_artifact": state["plan_artifact"],
+            "evaluation": evaluation,
+        }
+    )
+    return {"evaluation": evaluation, "report": report}
+
+
 async def _persist_governance(state: AssessmentGraphState) -> AssessmentGraphState:
     try:
         count = persist_assessment_control_evidence(
@@ -248,21 +277,25 @@ async def _persist_governance(state: AssessmentGraphState) -> AssessmentGraphSta
 def compile_assessment_graph():
     graph = StateGraph(AssessmentGraphState)
     graph.add_node("load_skill", _load_skill)
+    graph.add_node("plan_assessment", _plan)
     graph.add_node("build_document_context", _build_context)
     graph.add_node("gather_policy_history_and_evidence", _gather_context)
     graph.add_node("draft_assessment", _draft)
     graph.add_node("review_assessment", _review)
     graph.add_node("parse_report", _parse_report)
     graph.add_node("verify_threat_evidence", _verify_threat_evidence)
+    graph.add_node("evaluate_assessment", _evaluate)
     graph.add_node("persist_gate3_control_evidence", _persist_governance)
     graph.add_edge(START, "load_skill")
-    graph.add_edge("load_skill", "build_document_context")
+    graph.add_edge("load_skill", "plan_assessment")
+    graph.add_edge("plan_assessment", "build_document_context")
     graph.add_edge("build_document_context", "gather_policy_history_and_evidence")
     graph.add_edge("gather_policy_history_and_evidence", "draft_assessment")
     graph.add_edge("draft_assessment", "review_assessment")
     graph.add_edge("review_assessment", "parse_report")
     graph.add_edge("parse_report", "verify_threat_evidence")
-    graph.add_edge("verify_threat_evidence", "persist_gate3_control_evidence")
+    graph.add_edge("verify_threat_evidence", "evaluate_assessment")
+    graph.add_edge("evaluate_assessment", "persist_gate3_control_evidence")
     graph.add_edge("persist_gate3_control_evidence", END)
     return graph.compile()
 
@@ -275,7 +308,16 @@ async def run_assessment_graph(
     project_id: str | None = None,
     phase: str | None = None,
     skill_id: str | None = None,
+    task_contract: AgentTaskContract | None = None,
 ) -> AssessmentReport:
+    if task_contract is None:
+        from app.agent.task_contract import build_task_contract
+
+        task_contract = build_task_contract(
+            task_id=task_id,
+            parsed_documents=parsed_documents,
+            phase=phase,
+        )
     compiled = compile_assessment_graph()
     final_state = await compiled.ainvoke(
         {
@@ -285,6 +327,7 @@ async def run_assessment_graph(
             "project_id": project_id,
             "phase": phase,
             "skill_id": skill_id,
+            "task_contract": task_contract,
         }
     )
     return final_state["report"]

app/agent/graph/state_types.py

@@ -6,6 +6,11 @@
 from langchain_core.messages import BaseMessage
 from langgraph.graph.message import add_messages
 
+from app.models.agent_execution import (
+    AgentTaskContract,
+    EvaluationArtifact,
+    PlanArtifact,
+)
 from app.models.assessment import AssessmentReport
 from app.models.parser import ParsedDocument
 
@@ -58,3 +63,6 @@ class AssessmentGraphState(TypedDict, total=False):
     reviewed_raw: str
     report: AssessmentReport
     persisted_controls: int
+    task_contract: AgentTaskContract
+    plan_artifact: PlanArtifact
+    evaluation: EvaluationArtifact

app/agent/orchestrator.py

@@ -14,6 +14,7 @@
 from app.core.guardrails import UNTRUSTED_CONTENT_INSTRUCTION, wrap_untrusted_content
 from app.kb.service import get_kb_service
 from app.llm.base import invoke_llm
+from app.models.agent_execution import AgentTaskContract
 from app.models.assessment import (
     AssessmentReport,
     ComplianceGap,
@@ -119,6 +120,7 @@ async def run_assessment(
     project_id: str | None = None,
     phase: str | None = None,
     skill_id: str | None = None,
+    task_contract: AgentTaskContract | None = None,
 ) -> AssessmentReport:
     from app.agent.graph.assessment_graph import run_assessment_graph
 
@@ -129,6 +131,7 @@ async def run_assessment(
         project_id=project_id,
         phase=phase,
         skill_id=skill_id,
+        task_contract=task_contract,
     )
 
 

app/agent/task_contract.py

@@ -0,0 +1,190 @@
+"""Deterministic Task Contract, planning, and evaluation helpers."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from hashlib import sha256
+from uuid import UUID
+
+from app.models.agent_execution import (
+    AgentTaskContract,
+    EvaluationArtifact,
+    EvaluationCheck,
+    PlanArtifact,
+    PlanStep,
+    TaskInputReference,
+)
+from app.models.assessment import AssessmentReport
+from app.models.parser import ParsedDocument
+
+
+def build_task_contract(
+    *,
+    task_id: UUID,
+    parsed_documents: list[ParsedDocument],
+    phase: str | None,
+    created_at: datetime | None = None,
+) -> AgentTaskContract:
+    """Create the bounded contract shared by REST, MCP, and A2A assessments."""
+    inputs = tuple(_input_reference(document) for document in parsed_documents)
+    normalized_phase = phase or "auto"
+    risk_tier = (
+        "high"
+        if normalized_phase in {"deployment", "operations", "full_ssdlc"}
+        else "medium"
+    )
+    return AgentTaskContract(
+        task_id=task_id,
+        created_at=created_at or datetime.now(UTC),
+        goal=(
+            "Assess the submitted documents for security, compliance, and "
+            f"evidence gaps in the {normalized_phase} SSDLC phase."
+        ),
+        inputs=inputs,
+        allowed_paths=tuple(item.uri for item in inputs),
+        allowed_tools=(
+            "document.read",
+            "knowledge_base.search",
+            "llm.generate",
+            "evidence.verify",
+        ),
+        expected_outputs=("assessment_report.v2", "evaluation_artifact.v1"),
+        success_criteria=(
+            "The report satisfies the AssessmentReport schema.",
+            "Every supported citation resolves to an allowed input or trusted source.",
+            "Threat evidence is evaluated independently from report drafting.",
+            "The result remains pending until the configured human review completes.",
+        ),
+        risk_tier=risk_tier,
+        # Plan-first enforcement is a later M1 policy slice. Until then every
+        # task remains under the existing mandatory human-review boundary.
+        approval_mode="human_review",
+        retry_limit=2,
+        escalation_owner="security_reviewer",
+    )
+
+
+def build_plan_artifact(
+    contract: AgentTaskContract,
+    *,
+    skill_id: str | None,
+) -> PlanArtifact:
+    """Produce a visible plan without invoking tools or reading document content."""
+    return PlanArtifact(
+        task_id=contract.task_id,
+        created_at=datetime.now(UTC),
+        skill_id=skill_id,
+        steps=(
+            PlanStep(
+                id="plan-1",
+                phase="plan",
+                action="Validate the task contract and select the assessment skill.",
+                inputs=("agent_task_contract.v1",),
+                outputs=("plan_artifact.v1",),
+                success_check="Contract fields are valid and all inputs are in scope.",
+            ),
+            PlanStep(
+                id="act-1",
+                phase="act",
+                action="Build bounded document context and retrieve policy evidence.",
+                inputs=contract.allowed_paths,
+                outputs=("document_context", "policy_context", "history_context"),
+                success_check=(
+                    "Only allowed inputs and read-only retrieval tools are used."
+                ),
+            ),
+            PlanStep(
+                id="act-2",
+                phase="act",
+                action="Draft and independently review the structured assessment.",
+                inputs=("document_context", "policy_context", "history_context"),
+                outputs=("assessment_report.v2",),
+                success_check="The reviewer emits a schema-valid assessment report.",
+            ),
+            PlanStep(
+                id="evaluate-1",
+                phase="evaluate",
+                action="Verify evidence and apply deterministic success checks.",
+                inputs=("assessment_report.v2",),
+                outputs=("evaluation_artifact.v1",),
+                success_check=(
+                    "Required checks pass or the result is marked for review."
+                ),
+            ),
+        ),
+    )
+
+
+def evaluate_assessment_report(
+    report: AssessmentReport,
+    contract: AgentTaskContract,
+) -> EvaluationArtifact:
+    """Evaluate generated output with deterministic checks outside the drafter."""
+    schema_valid = True
+    schema_details = "AssessmentReport schema validation passed."
+    try:
+        AssessmentReport.model_validate(report.model_dump())
+    except ValueError as exc:
+        schema_valid = False
+        schema_details = f"AssessmentReport schema validation failed: {exc}"
+
+    finding_count = len(report.risk_items) + len(report.compliance_gaps)
+    evidence_present = finding_count == 0 or bool(report.sources)
+    threats = report.threat_model.threats if report.threat_model else []
+    threats_verified = all(threat.verification is not None for threat in threats)
+
+    checks = (
+        EvaluationCheck(
+            name="schema_valid",
+            passed=schema_valid,
+            details=schema_details,
+        ),
+        EvaluationCheck(
+            name="task_identity_matches",
+            passed=report.task_id == str(contract.task_id),
+            details="Report task ID must match the immutable task contract.",
+        ),
+        EvaluationCheck(
+            name="required_output_present",
+            passed=bool(report.summary.strip()) and report.status != "failed",
+            details="A non-empty, non-failed assessment report is required.",
+        ),
+        EvaluationCheck(
+            name="finding_evidence_present",
+            passed=evidence_present,
+            required=False,
+            details="Findings should include at least one structured source citation.",
+        ),
+        EvaluationCheck(
+            name="threat_evidence_verified",
+            passed=threats_verified,
+            details="Every generated threat must pass through the evidence critic.",
+        ),
+    )
+    required_failed = any(not check.passed for check in checks if check.required)
+    advisory_failed = any(not check.passed for check in checks if not check.required)
+    if required_failed:
+        outcome = "failed"
+    elif advisory_failed:
+        outcome = "needs_review"
+    else:
+        outcome = "passed"
+    return EvaluationArtifact(
+        task_id=contract.task_id,
+        created_at=datetime.now(UTC),
+        outcome=outcome,
+        checks=checks,
+    )
+
+
+def _input_reference(document: ParsedDocument) -> TaskInputReference:
+    content = (
+        document.content if isinstance(document.content, str) else str(document.content)
+    )
+    digest = document.metadata.file_hash or sha256(content.encode("utf-8")).hexdigest()
+    return TaskInputReference(
+        uri=f"document://{digest}",
+        filename=document.metadata.filename,
+        media_type=document.metadata.type,
+        sha256=digest,
+    )

app/models/__init__.py

@@ -1,3 +1,11 @@
+from .agent_execution import (
+    AgentTaskContract,
+    EvaluationArtifact,
+    EvaluationCheck,
+    PlanArtifact,
+    PlanStep,
+    TaskInputReference,
+)
 from .assessment import (
     AssessmentReport,
     AssessmentTaskCreated,
@@ -38,6 +46,7 @@
 __all__ = [
     "AgentIntegrationStatus",
     "AgentProtocolEndpoint",
+    "AgentTaskContract",
     "AssessmentReport",
     "AssessmentTaskCreated",
     "AssessmentTaskResult",
@@ -48,12 +57,16 @@
     "ControlInstance",
     "CrossPhaseRef",
     "DreadScore",
+    "EvaluationArtifact",
+    "EvaluationCheck",
     "GateSubmission",
     "GovernanceAuditLog",
     "OrgFrameworkConfig",
     "ParsedDocument",
     "PolicyDocument",
     "PolicyEmbedding",
+    "PlanArtifact",
+    "PlanStep",
     "Project",
     "PromptAuditLog",
     "QuestionInstance",
@@ -66,6 +79,7 @@
     "SubAgentRun",
     "Threat",
     "ThreatModel",
+    "TaskInputReference",
     "TrackedRemediation",
     "User",
     "Vulnerability",