diff --git a/docs/engineering/codex_dgx_worker.md b/docs/engineering/codex_dgx_worker.md index 914429b84..bb8e7152d 100644 --- a/docs/engineering/codex_dgx_worker.md +++ b/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. diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py index 3fb0dc892..5ebc59a75 100755 --- a/scripts/codex_von_worker.py +++ b/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, diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index 65f8b667a..dcfa907ca 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/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( diff --git a/src/backend/services/task_management_service.py b/src/backend/services/task_management_service.py index 3bba807ce..543f7e0ef 100644 --- a/src/backend/services/task_management_service.py +++ b/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 } diff --git a/src/backend/utils/task_execution_preferences.py b/src/backend/utils/task_execution_preferences.py new file mode 100644 index 000000000..8668666b0 --- /dev/null +++ b/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 diff --git a/tests/backend/test_codex_von_worker.py b/tests/backend/test_codex_von_worker.py index 8ae841c47..6dbde73e0 100644 --- a/tests/backend/test_codex_von_worker.py +++ b/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( diff --git a/tests/backend/test_coding_task_creation_recovery.py b/tests/backend/test_coding_task_creation_recovery.py index 427cbf99b..d78ea5c5e 100644 --- a/tests/backend/test_coding_task_creation_recovery.py +++ b/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 ): diff --git a/tests/test_codex_von_release.py b/tests/test_codex_von_release.py index 281451750..5a497f464 100644 --- a/tests/test_codex_von_release.py +++ b/tests/test_codex_von_release.py @@ -39,6 +39,11 @@ def source(tmp_path): service = source / "src/backend/services/task_management_service.py" service.parent.mkdir(parents=True) service.write_text("RELEASE = 'old'\n") + preferences = source / "src/backend/utils/task_execution_preferences.py" + preferences.parent.mkdir(parents=True) + preferences.write_bytes( + (repo / "src/backend/utils/task_execution_preferences.py").read_bytes() + ) git(source, "add", ".") git(source, "commit", "-qm", "old fixture") old = git(source, "rev-parse", "HEAD") @@ -75,7 +80,11 @@ def test_complete_release_switch_preserves_loaded_files_and_rollback(source, tmp import codex_von_worker as worker worker.bind_backend_root(root) from src.backend.services import task_management_service as tasks -print(json.dumps([worker.__file__, tasks.__file__, tasks.RELEASE])) +from src.backend.utils import task_execution_preferences as preferences +settings = worker.resolve_execution_settings({}, { + 'requested_model': 'gpt-6-astra', 'requested_reasoning_effort': 'xhigh'}) +print(json.dumps([worker.__file__, tasks.__file__, tasks.RELEASE, + preferences.__file__, settings])) """ result = json.loads( subprocess.check_output( @@ -86,6 +95,13 @@ def test_complete_release_switch_preserves_loaded_files_and_rollback(source, tmp str(new_path / "scripts/codex_von_worker.py"), str(new_path / "src/backend/services/task_management_service.py"), "new", + str(new_path / "src/backend/utils/task_execution_preferences.py"), + { + "model": "gpt-6-astra", + "model_source": "task", + "reasoning_effort": "xhigh", + "reasoning_effort_source": "task", + }, ] assert releases.prepare(source, root, new) == new_path releases.activate(root, old_path)