Validate coding-task execution preferences before writes and launch

docs/engineering/codex_dgx_worker.md

@@ -28,8 +28,20 @@ conversation text. Changes apply at the next launch, not midway through a run.
 
 The launch passes both resolved values explicitly to Codex and retains their
 values and sources in `execution_settings` in the run context/checkpoint and
-result message. Unsupported selections fail visibly through Codex; the worker
-does not substitute another model or reasoning level. The standing Sol block
+result message. Canonical creation/updates and worker launch share a narrow
+token validator: agent concept IDs, whitespace-containing values and the
+observed `CodexDGX`/`Astra` display-label mistakes are rejected as model IDs.
+`extra_high`/`extra-high` reasoning returns guidance to retry with `xhigh`,
+preserving the explicit choice. Rejected writes do not partially apply other
+task fields. No label is automatically mapped to a model or default.
+
+This is not an availability catalogue: other exact model and reasoning tokens
+are preserved, including future provider-supported values. The installed Codex
+protocol describes reasoning as a model-advertised string; the
+[public configuration reference](https://learn.chatgpt.com/docs/config-file/config-reference)
+also documents `xhigh`. Model/account availability and model-specific effort
+compatibility remain Codex's responsibility and can still fail visibly there.
+The worker does not substitute another model or reasoning level. The standing Sol block
 also applies to task overrides. Reply-only inbox runs use operator defaults
 because task selection happens within that reply.
 

scripts/codex_von_worker.py

@@ -58,6 +58,11 @@ def fingerprint(value):
 
 def resolve_execution_settings(config, inputs):
     """Resolve task preferences without substituting another requested model."""
+    # Import after the controller binds its coherent backend release.
+    from src.backend.utils.task_execution_preferences import (
+        normalise_task_execution_preference,
+    )
+
     resolved = {}
     for name, default in (("model", "gpt-6-astra"), ("reasoning_effort", "medium")):
         requested = inputs.get("requested_" + name)
@@ -66,15 +71,14 @@ def resolve_execution_settings(config, inputs):
         )
         value = requested if requested is not None else configured
         value = default if value is None else value
-        if (
-            not isinstance(value, str)
-            or not value.strip()
-            or any(c.isspace() for c in value.strip())
-        ):
+        value = normalise_task_execution_preference(
+            value, field_name="requested_" + name
+        )
+        if value is None:
             raise ValueError(
                 f"Invalid {name}: specify one exact model or reasoning level"
             )
-        resolved[name] = value.strip()
+        resolved[name] = value
         resolved[name + "_source"] = (
             "task" if requested is not None else "worker_default"
         )
@@ -979,7 +983,6 @@ def main():
         parser.error("Backfill is bounded to 20 selected pairs")
     os.umask(0o077)
     config = json.loads(Path(options.config).read_text())
-    resolve_execution_settings(config, {})
     state_root = Path(config["state_root"])
     state_root.mkdir(parents=True, exist_ok=True)
     with (state_root / "worker.lock").open("a") as lock:
@@ -989,6 +992,7 @@ def main():
             print('{"outcome":"already_running"}')
             return
         backend_root = bind_backend_root(options.backend_root)
+        resolve_execution_settings(config, {})
         from src.backend.integrations.internal_mcp.gateway import (
             INTERNAL_MCP_TRUSTED_LOCAL_OPERATOR_SOURCE,
             bind_internal_mcp_actor_context_source,

src/backend/integrations/internal_mcp/catalogue.py

@@ -35261,11 +35261,19 @@ def _task_create(**kwargs):
     evidence = kwargs.get("evidence")
     notes = kwargs.get("notes")
     reference_code = kwargs.get("reference_code")
-    requested_settings = {
-        key: (value.strip() or None) if isinstance(value, str) else value
-        for key in ("requested_model", "requested_reasoning_effort")
-        if (value := kwargs.get(key)) is not None
-    }
+    from ...utils.task_execution_preferences import normalise_task_execution_preference
+
+    try:
+        requested_settings = {
+            key: normalise_task_execution_preference(value, field_name=key)
+            for key in ("requested_model", "requested_reasoning_effort")
+            if (value := kwargs.get(key)) is not None
+        }
+    except ValueError as exc:
+        return {
+            **make_error_response("INVALID_DATA", str(exc)),
+            "changed": False,
+        }
     request_id = str(kwargs.get("request_id") or "").strip()
     idempotency_key = str(kwargs.get("idempotency_key") or "").strip()
 
@@ -45032,8 +45040,12 @@ def _build_default_catalogue_task_and_workflow_definitions() -> List[MethodDefin
                 "task categories, source semantics, and richer task-detail fields. "
                 "For coding-agent work, preserve a user-specified model and reasoning level in "
                 "requested_model (exact model ID, e.g. gpt-6-astra) and "
-                "requested_reasoning_effort (e.g. medium or high). Omit either to inherit "
-                "the coding worker default. These fields do not execute the task."
+                "requested_reasoning_effort (e.g. high or xhigh; extra-high is xhigh). "
+                "Agent names/IDs belong in assignee_concept_id, never requested_model; "
+                "use an exact model ID, not a display label such as Astra. Omit a "
+                "preference to inherit the worker default only if none was requested. "
+                "Malformed preferences are rejected before creation; correct and retry "
+                "without dropping the user's choice. These fields do not execute the task."
             ),
         ),
         MethodDefinition(

src/backend/services/task_management_service.py

@@ -36,6 +36,7 @@
 from ..utils.concept_id_utils import (
     ensure_v_concept_prefix,
 )
+from ..utils.task_execution_preferences import normalise_task_execution_preference
 from .effort_unit_ontology_service import (
     ensure_effort_unit_ontology,
     extract_effort_unit_type_ids,
@@ -427,6 +428,13 @@ def _relationship_id_list(value: Any) -> list[str]:
     return ordered
 
 
+def _normalise_execution_preference(value: Any, *, field_name: str) -> str | None:
+    try:
+        return normalise_task_execution_preference(value, field_name=field_name)
+    except ValueError as exc:
+        raise InvalidTaskDataError(str(exc)) from exc
+
+
 def _normalise_optional_text(value: Any, *, field_name: str) -> str | None:
     if value is None:
         return None
@@ -1398,10 +1406,10 @@ def create_task(
         reference_code,
         field_name="reference_code",
     )
-    requested_model = _normalise_optional_text(
+    requested_model = _normalise_execution_preference(
         requested_model, field_name="requested_model"
     )
-    requested_reasoning_effort = _normalise_optional_text(
+    requested_reasoning_effort = _normalise_execution_preference(
         requested_reasoning_effort, field_name="requested_reasoning_effort"
     )
     agent_creation_fingerprint = _normalise_optional_text(
@@ -4760,7 +4768,7 @@ def update_task_fields(
     existing_task = _build_task_response(task_doc)
     # Validate these settings before changing any of the requested task fields.
     execution_settings = {
-        key: _normalise_optional_text(fields[key], field_name=key)
+        key: _normalise_execution_preference(fields[key], field_name=key)
         for key in ("requested_model", "requested_reasoning_effort")
         if key in fields
     }

src/backend/utils/task_execution_preferences.py

@@ -0,0 +1,51 @@
+"""Validate task execution tokens without choosing a model or provider policy.
+
+This is a narrow input contract, not a catalogue of available models/efforts.
+Codex advertises reasoning levels dynamically; preserve exact future/provider
+tokens. Reject the demonstrated task-field mix-ups, never infer a replacement.
+"""
+
+from __future__ import annotations
+
+
+def normalise_task_execution_preference(
+    value: object, *, field_name: str
+) -> str | None:
+    if value is None:
+        return None
+    if not isinstance(value, str):
+        raise ValueError(f"{field_name} must be a string")
+    value = value.strip()
+    if not value:
+        return None
+
+    if field_name == "requested_model":
+        # Exact observed display labels, not a model-family allowlist or a
+        # lexical guess about arbitrary future model IDs.
+        if (
+            value.casefold() in {"codexdgx", "astra"}
+            or value.startswith("#V#")
+            or any(c.isspace() or ord(c) < 32 for c in value)
+        ):
+            raise ValueError(
+                "requested_model requires an exact model ID (for example "
+                "gpt-6-astra), not an agent concept ID or display label. Put the "
+                "coding agent in assignee_concept_id; supply the user's exact "
+                "model ID, or omit/clear requested_model to inherit the worker "
+                "default only when no model was requested."
+            )
+    elif field_name == "requested_reasoning_effort":
+        if value.casefold() in {"extra_high", "extra-high", "extra high"}:
+            raise ValueError(
+                "requested_reasoning_effort requires the exact level xhigh for "
+                "extra-high reasoning. Retry with xhigh to preserve that choice; "
+                "do not drop it or substitute high/the worker default."
+            )
+        if any(c.isspace() or ord(c) < 32 for c in value):
+            raise ValueError(
+                "requested_reasoning_effort requires one exact model-supported "
+                "reasoning token (for example high or xhigh)."
+            )
+    else:
+        raise ValueError(f"Unknown task execution preference: {field_name}")
+    return value

tests/backend/test_codex_von_worker.py

@@ -75,6 +75,13 @@ def test_empty_poll_has_no_model_or_message(config, monkeypatch):
         ({}, ("gpt-6-astra", "medium")),
         ({"requested_model": "gpt-5.6-terra"}, ("gpt-5.6-terra", "medium")),
         ({"requested_reasoning_effort": "high"}, ("gpt-6-astra", "high")),
+        ({"requested_reasoning_effort": "xhigh"}, ("gpt-6-astra", "xhigh")),
+        ({"requested_reasoning_effort": "max"}, ("gpt-6-astra", "max")),
+        ({"requested_reasoning_effort": "ultra"}, ("gpt-6-astra", "ultra")),
+        (
+            {"requested_model": "provider/Model-Preview_vNext:2026"},
+            ("provider/Model-Preview_vNext:2026", "medium"),
+        ),
         (
             {"requested_model": None, "requested_reasoning_effort": None},
             ("gpt-6-astra", "medium"),
@@ -92,14 +99,29 @@ def test_task_model_settings_inherit_independently(requested, expected):
         )
 
 
-def test_forbidden_task_model_is_reported_without_running_codex(config, monkeypatch):
+@pytest.mark.parametrize(
+    "preferences,error",
+    [
+        ({"requested_model": "gpt-5.6-sol"}, "Sol-family"),
+        ({"requested_model": "Codex DGX"}, "exact model ID"),
+        ({"requested_model": "DGX Codex"}, "exact model ID"),
+        ({"requested_model": "#V#codex_dgx"}, "assignee_concept_id"),
+        ({"requested_model": "CodexDGX"}, "exact model ID"),
+        ({"requested_model": "Astra"}, "exact model ID"),
+        ({"requested_reasoning_effort": "extra_high"}, "xhigh"),
+        ({"requested_reasoning_effort": "extra-high"}, "xhigh"),
+    ],
+)
+def test_invalid_task_preferences_are_reported_without_running_codex(
+    config, monkeypatch, preferences, error
+):
     monkeypatch.setattr(
         worker.subprocess, "run", lambda *a, **kw: pytest.fail("model invoked")
     )
     reports = []
     api = SimpleNamespace(
         pending=lambda: [task(config)],
-        inputs=lambda _: {"requested_model": "gpt-5.6-sol"},
+        inputs=lambda _: preferences,
         conversation=lambda _: {"available": False},
         send=lambda *a: "#V#started",
         tasks=SimpleNamespace(update_task_status=lambda *a, **kw: None),
@@ -110,7 +132,7 @@ def test_forbidden_task_model_is_reported_without_running_codex(config, monkeypa
     )
     worker.tick(config, api, 0)
     assert reports[0]["status"] == "blocked"
-    assert "Sol-family" in reports[0]["summary"]
+    assert error in reports[0]["summary"]
 
 
 @pytest.mark.parametrize(

tests/backend/test_coding_task_creation_recovery.py

@@ -38,6 +38,10 @@
 def native_store(monkeypatch):
     db = mongomock.MongoClient()["isolated_coding_task_acceptance_2750"]
     db.concepts.create_index("concept_id", unique=True)
+    monkeypatch.setattr("src.backend.db.mongo_client.get_db", lambda: db)
+    monkeypatch.setattr(
+        "src.backend.db.mongo_client.get_concepts_collection", lambda: db.concepts
+    )
     monkeypatch.setattr(ConceptsRepository, "collection", lambda: db.concepts)
     # The fixture does not replicate the ontology catalogue used for target
     # visibility. Actor/org query filtering and the assignment checks stay real.
@@ -268,6 +272,92 @@ def test_create_schema_exposes_preferences_without_exposing_trusted_creator(gate
     assert "created_by_concept_id" not in schema["properties"]
 
 
+@pytest.mark.parametrize(
+    "field,value,hint",
+    [
+        ("requested_model", "DGX Codex", "assignee_concept_id"),
+        ("requested_model", "Codex DGX", "assignee_concept_id"),
+        ("requested_model", "#V#codex_dgx", "assignee_concept_id"),
+        ("requested_model", "CodexDGX", "exact model ID"),
+        ("requested_model", "Astra", "exact model ID"),
+        ("requested_reasoning_effort", "extra_high", "xhigh"),
+        ("requested_reasoning_effort", "extra-high", "xhigh"),
+    ],
+)
+def test_malformed_preferences_reject_before_writes_and_corrected_retry_is_unique(
+    native_store, gateway, field, value, hint
+):
+    with override_current_actor("#V#alice", "#V#lab"):
+        invalid = payload(gateway, **{field: value})
+        rejected = gateway.invoke("task_create", invalid).payload
+        assert rejected["error_code"] == "INVALID_DATA", rejected
+        assert hint in rejected["error"]
+        assert rejected["changed"] is False
+        assert native_store.concepts.count_documents({}) == 0
+        assert native_store.relations.count_documents({}) == 0
+
+        # The canonical service also protects non-MCP task creation.
+        with pytest.raises(tasks.InvalidTaskDataError, match=hint):
+            tasks.create_task(
+                title="Invalid fixture", description="Never launch", **{field: value}
+            )
+        assert native_store.concepts.count_documents({}) == 0
+
+        corrected = payload(gateway, requested_reasoning_effort="xhigh")
+        created = gateway.invoke("task_create", corrected).payload
+        assert created["success"], created
+        task_id = created["task_concept_id"]
+        repeated = gateway.invoke("task_create", corrected).payload
+        assert repeated["success"] and repeated["task_concept_id"] == task_id
+        assert repeated["changed"] is False
+        before = tasks.get_task(task_id)
+        assert before["requested_model"] == "gpt-6-astra"
+        assert before["requested_reasoning_effort"] == "xhigh"
+        assert before["assignee_concept_id"] == "#V#worker"
+
+        # Reject the complete edit before status/notes or preferences can change.
+        rejected_update = gateway.invoke(
+            "task_update_fields",
+            {
+                "task_concept_id": task_id,
+                "fields": {
+                    "status": "completed",
+                    "notes": "Must not persist",
+                    field: value,
+                },
+            },
+        ).payload
+        assert rejected_update["error_code"] == "INVALID_DATA", rejected_update
+        assert hint in rejected_update["error"]
+        assert tasks.get_task(task_id) == before
+        assert native_store.concepts.count_documents({}) == 1
+
+
+def test_exact_future_execution_tokens_are_preserved(native_store, gateway):
+    # Availability belongs to the execution provider, not a frozen task allowlist.
+    model = "provider/Model-Preview_vNext:2026"
+    effort = "future_effort"
+    with override_current_actor("#V#alice", "#V#lab"):
+        result = gateway.invoke(
+            "task_create",
+            payload(gateway, requested_model=model, requested_reasoning_effort=effort),
+        ).payload
+        assert result["success"], result
+        task_id = result["task_concept_id"]
+        task = tasks.get_task(task_id)
+        assert task["requested_model"] == model
+        assert task["requested_reasoning_effort"] == effort
+        tasks.update_task_fields(
+            task_id, fields={"requested_reasoning_effort": "ultra"}
+        )
+        task = tasks.get_task(task_id)
+        worker = load_worker(
+            Path(__file__).resolve().parents[2] / "scripts/codex_von_worker.py"
+        )
+        resolved = worker.resolve_execution_settings({}, task)
+        assert resolved["model"] == model and resolved["reasoning_effort"] == "ultra"
+
+
 def test_creation_failed_parent_update_and_assignment_recovery_answer(
     native_store, gateway, monkeypatch
 ):