Reconcile verified task creation recovery before outcome reporting

src/backend/services/adaptive_turn_service.py

@@ -4764,6 +4764,100 @@ def _effect_postcondition_identity(
     return hashlib.sha256(_json_bytes(identity)).hexdigest(), identity
 
 
+def _reconcile_task_creation_attempts(
+    tool_invocations: Sequence[Mapping[str, Any]],
+    projected: dict[str, dict[str, Any]],
+) -> None:
+    """Reconcile a rejected create with a verified record for the same request.
+
+    A corrected invalid type changes the handler fingerprint, so neither an
+    exact effect ID nor a task ID exists to join the rejected attempt. Compare
+    the remaining creation arguments instead; assignment may be completed by
+    a later verified field update on the returned task. This is an evidence
+    projection only, and never changes the original attempt receipt.
+    """
+    assignment_fields = ("assignee_concept_id", "report_to_concept_id")
+
+    def arguments_for(invocation: Mapping[str, Any]) -> dict[str, Any]:
+        arguments = dict(invocation.get("effective_arguments") or {})
+        for alias, field in (
+            ("assignee_id", "assignee_concept_id"),
+            ("reports_to_concept_id", "report_to_concept_id"),
+            ("task_type_id", "task_type_ids"),
+        ):
+            if alias in arguments:
+                arguments.setdefault(field, arguments.pop(alias))
+        if not arguments.get("assignee_concept_id") and arguments.get(
+            "created_by_concept_id"
+        ):
+            arguments["assignee_concept_id"] = arguments["created_by_concept_id"]
+        arguments.setdefault("report_to_concept_id", None)
+        return arguments
+
+    for index, invocation in enumerate(tool_invocations):
+        if invocation.get("execution_method", invocation.get("tool")) != "task_create":
+            continue
+        state = projected.get(invocation.get("effect_id"), {})
+        if (
+            state.get("effect_status") not in {"failed", "not_started"}
+            or state.get("changed") is not False
+        ):
+            continue
+        expected = arguments_for(invocation)
+        if not expected.get("title") or not expected.get("description"):
+            continue
+        error_code, error = _effect_failure_fact(state)
+        corrected_invalid_type = (
+            error_code == "INVALID_DATA"
+            and "unknown task type" in str(error or "").lower()
+        )
+        for later_index in range(index + 1, len(tool_invocations)):
+            later = tool_invocations[later_index]
+            if later.get("execution_method", later.get("tool")) != "task_create":
+                continue
+            later_state = projected.get(later.get("effect_id"), {})
+            receipt = later_state.get("canonical_readback") or {}
+            task_id = receipt.get("task_concept_id")
+            if (
+                later_state.get("effect_status") != "succeeded"
+                or receipt.get("verified") is not True
+                or not task_id
+            ):
+                continue
+            observed_arguments = arguments_for(later)
+            excluded = set(assignment_fields)
+            if corrected_invalid_type:
+                excluded.add("task_type_ids")
+            if {k: v for k, v in expected.items() if k not in excluded} != {
+                k: v for k, v in observed_arguments.items() if k not in excluded
+            }:
+                continue
+            fields = dict(receipt.get("task_fields") or {})
+            recovery_id = later["effect_id"]
+            for update in tool_invocations[later_index + 1 :]:
+                update_state = projected.get(update.get("effect_id"), {})
+                update_receipt = update_state.get("canonical_readback") or {}
+                if (
+                    update.get("execution_method", update.get("tool"))
+                    == "task_update_fields"
+                    and update_state.get("effect_status") == "succeeded"
+                    and update_receipt.get("verified") is True
+                    and update_receipt.get("task_concept_id") == task_id
+                ):
+                    fields.update(update_receipt.get("task_fields") or {})
+                    recovery_id = update["effect_id"]
+            if any(
+                field in expected and fields.get(field) != expected[field]
+                for field in assignment_fields
+            ):
+                continue
+            state["recovered_by_effect_id"] = recovery_id
+            state["recovery_status"] = "succeeded"
+            state["reconciliation_basis"] = "later_verified_requested_task_record"
+            state["recovered_task_concept_id"] = task_id
+            break
+
+
 def _reconcile_task_update_attempts(
     tool_invocations: Sequence[Mapping[str, Any]],
     projected: dict[str, dict[str, Any]],
@@ -4872,6 +4966,7 @@ def _reconcile_effect_attempts_by_postcondition(
             if success[0] > index
             else "prior_exact_postcondition_success"
         )
+    _reconcile_task_creation_attempts(tool_invocations, projected)
     _reconcile_task_update_attempts(tool_invocations, projected)
     return projected
 
@@ -5880,6 +5975,11 @@ def _build_effect_outcome_report(
             effect_status in {"failed", "partial", "indeterminate", "not_started"}
             or state.get("changed") is True
             or (effect_status == "succeeded" and state.get("changed") is None)
+            or (
+                effect_status == "succeeded"
+                and isinstance(state.get("canonical_readback"), Mapping)
+                and state["canonical_readback"].get("verified") is True
+            )
             or state.get("outcome_resolved") is True
         ):
             continue

tests/backend/test_adaptive_turn_service.py

@@ -16441,3 +16441,266 @@ def reader(*, close_if_empty=False):
         client.calls[1]["context"][-1]["content"]
         == "Use this evidence in a shorter answer"
     )
+
+
+@pytest.mark.parametrize(
+    "scenario", ["direct", "recovery", "assign", "replay", "partial", "partial_replay", "failure"]
+)
+def test_task_creation_requested_outcome_reaches_final_response(scenario: str) -> None:
+    from src.backend.integrations.internal_mcp.schemas import make_error_response
+    from src.backend.services.turn_execution_record_service import (
+        _build_tool_authored_mutation_effects,
+    )
+
+    task_id = "#V#task_agent_verified"
+    desired = {
+        "title": "Repair task reporting",
+        "description": "Verify and deliver the repair.",
+        "assignee_concept_id": "#V#codex_dgx",
+    }
+    receipt = {
+        "verified": True,
+        "status": "verified",
+        "task_concept_id": task_id,
+        "task_fields": {"assignee_concept_id": "#V#codex_dgx"},
+    }
+
+    def handler(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
+        if (
+            arguments.get("task_type_id") == "#V#task_specification"
+            or name == "other_effect"
+        ):
+            return make_error_response("INVALID_DATA", "Unknown task type")
+        actual_receipt = {
+            **receipt,
+            "task_fields": {
+                "assignee_concept_id": arguments.get(
+                    "assignee_concept_id", "#V#codex_dgx"
+                )
+            },
+        }
+        return {
+            "success": True,
+            "effect_status": "succeeded",
+            "changed": scenario not in {"replay", "partial_replay"},
+            "idempotent_replay": scenario in {"replay", "partial_replay"},
+            "task_concept_id": task_id,
+            "canonical_readback": actual_receipt,
+        }
+
+    catalogue = MethodCatalogue()
+    for name in ("task_create", "task_update_fields", "other_effect"):
+        catalogue.register(
+            MethodDefinition(
+                name=name,
+                handler=lambda _name=name, **kwargs: handler(_name, kwargs),
+                input_schema=Schema(allow_unknown=True),
+                category="write",
+                ordinary_turn_effect=True,
+            )
+        )
+    gateway = InternalMCPGateway(
+        catalogue=catalogue,
+        transport=InternalMCPTransport(write_timeout_sec=1),
+        enabled=True,
+    )
+    calls = []
+
+    def call(name: str, arguments: dict[str, Any]) -> None:
+        calls.append(
+            LLMResponse(
+                text_response="",
+                tool_calls=[
+                    ToolCall(
+                        tool_name="turn_invoke_capability",
+                        call_id=f"call-{len(calls)}",
+                        payload={"name": name, "arguments": arguments},
+                    )
+                ],
+            )
+        )
+
+    if scenario != "direct":
+        call("task_create", {**desired, "task_type_id": "#V#task_specification"})
+    if scenario != "failure":
+        call(
+            "task_create",
+            {
+                **desired,
+                **(
+                    {"assignee_concept_id": "#V#person"} if scenario == "assign" else {}
+                ),
+            },
+        )
+    if scenario == "assign":
+        call(
+            "task_update_fields",
+            {
+                "task_concept_id": task_id,
+                "fields": {"assignee_concept_id": "#V#codex_dgx"},
+            },
+        )
+    if scenario in {"partial", "partial_replay"}:
+        call("other_effect", {})
+    answer = (
+        f"Created task {task_id} and assigned it to #V#codex_dgx."
+        if scenario != "failure"
+        else "Task creation failed; no task was created."
+    )
+    client = _SequenceClient(*calls, LLMResponse(text_response=answer))
+    result = execute_adaptive_turn(
+        gateway=gateway,
+        prompt="Create and delegate the repair task.",
+        context=[],
+        llm_client=client,
+        model="test-model",
+        user_namespace="#V#person@org",
+        user_concept_id="#V#person",
+        org_concept_id="#V#org",
+        turn_id=f"task-outcome-{scenario}",
+        turn_budget_seconds=10,
+        final_synthesis_reserve_seconds=2,
+    )
+    if scenario in {"direct", "recovery", "assign", "replay"}:
+        assert result.terminal_status == "completed"
+        assert result.response_authority == "model"
+        assert result.response_text == answer
+        assert "Effect outcome report" not in result.response_text
+        if scenario != "direct":
+            assert result.tool_invocations[0]["recovery_status"] == "succeeded"
+            assert result.tool_invocations[0]["effect_status"] == "failed"
+        effects = _build_tool_authored_mutation_effects(
+            tool_invocations=result.tool_invocations,
+        )
+        assert effects
+        assert all(effect["status"] == "satisfied" for effect in effects)
+    else:
+        assert result.terminal_status == (
+            "effect_failed" if scenario == "failure" else "effect_partially_completed"
+        )
+        reports = [
+            row
+            for row in result.aux_llm_calls
+            if row.get("type") == "adaptive_turn_effect_outcome_report"
+        ]
+        report = reports[-1]
+        if scenario in {"partial", "partial_replay"}:
+            assert task_id in result.response_text
+            failed = [
+                fact
+                for fact in report["facts"]
+                if not fact.get("recovered_by_effect_id")
+                and fact["effect_status"] == "failed"
+            ]
+            assert [fact["tool"] for fact in failed] == ["other_effect"]
+            task_outcomes = [
+                fact["outcome"]
+                for fact in report["narration_context"]["operation_outcomes"]
+                if fact["operation"] == "Task create"
+            ]
+            assert task_outcomes
+            assert set(task_outcomes) <= {
+                "succeeded", "recovered by a later successful retry"
+            }
+        else:
+            assert not any(
+                fact.get("canonical_readback_verified") for fact in report["facts"]
+            )
+
+
+@pytest.mark.parametrize(
+    "difference",
+    [
+        "none",
+        "title",
+        "description",
+        "organisation_concept_id",
+        "requested_model",
+        "assignee_concept_id",
+        "unverified",
+        "changed",
+        "other_type_error",
+        "later_reassignment",
+    ],
+)
+def test_task_create_reconciliation_requires_matching_verified_requested_object(
+    difference: str,
+) -> None:
+    arguments = {
+        "title": "Requested work",
+        "description": "Exact scope",
+        "organisation_concept_id": "#V#org",
+        "assignee_concept_id": "#V#codex_dgx",
+        "requested_model": "test-model",
+    }
+    later_arguments = dict(arguments)
+    fields = {"assignee_concept_id": "#V#codex_dgx"}
+    if difference in arguments:
+        later_arguments[difference] = "different"
+    if difference == "assignee_concept_id":
+        fields["assignee_concept_id"] = "different"
+    invocations = [
+        {
+            "effect_id": "failed",
+            "tool": "task_create",
+            "effective_arguments": {
+                **arguments,
+                "task_type_id": "#V#task_specification",
+            },
+        },
+        {
+            "effect_id": "created",
+            "tool": "task_create",
+            "effective_arguments": later_arguments,
+        },
+    ]
+    snapshot = {
+        "failed": {
+            "effect_status": "failed",
+            "changed": difference == "changed",
+            "failure_fact": {
+                "error_code": "INVALID_DATA",
+                "error": (
+                    "Other validation failure"
+                    if difference == "other_type_error"
+                    else "Unknown task type"
+                ),
+            },
+        },
+        "created": {
+            "effect_status": "succeeded",
+            "changed": True,
+            "canonical_readback": {
+                "verified": difference != "unverified",
+                "task_concept_id": "#V#created",
+                "task_fields": fields,
+            },
+        },
+    }
+    if difference == "later_reassignment":
+        invocations.append(
+            {
+                "effect_id": "reassigned",
+                "tool": "task_update_fields",
+                "effective_arguments": {
+                    "task_concept_id": "#V#created",
+                    "fields": {"assignee_concept_id": "#V#other"},
+                },
+            }
+        )
+        snapshot["reassigned"] = {
+            "effect_status": "succeeded",
+            "changed": True,
+            "canonical_readback": {
+                "verified": True,
+                "task_concept_id": "#V#created",
+                "task_fields": {"assignee_concept_id": "#V#other"},
+            },
+        }
+    reconciled = _reconcile_effect_attempts_by_postcondition(
+        tool_invocations=invocations, effect_snapshot=snapshot
+    )
+    assert (reconciled["failed"].get("recovery_status") == "succeeded") is (
+        difference == "none"
+    )
+    assert "recovery_status" not in snapshot["failed"]