diff --git a/src/backend/services/adaptive_turn_service.py b/src/backend/services/adaptive_turn_service.py index daf0f275..150e64d2 100644 --- a/src/backend/services/adaptive_turn_service.py +++ b/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 diff --git a/tests/backend/test_adaptive_turn_service.py b/tests/backend/test_adaptive_turn_service.py index cf312828..0ca21d91 100644 --- a/tests/backend/test_adaptive_turn_service.py +++ b/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"]