From b933cf52c8c67dce8d6f03facecfcfbda7b17861 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 22 Aug 2026 23:02:41 +0200 Subject: [PATCH] Fail closed on invalid Glas profiles Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028de-e2c8-7732-8521-46a7fc5db82f --- .env.example | 5 + WORK-RECORDS.md | 2 +- docs/ops-run-queue.md | 25 +++- src/activity_core/activities.py | 40 +++++- src/activity_core/definition_parser.py | 74 ++++++++++ src/activity_core/models.py | 6 + src/activity_core/rules/actions.py | 21 +++ tests/rules/test_actions.py | 32 +++++ tests/test_glas_profile_selection.py | 8 ++ tests/test_issue_sink.py | 131 ++++++++++++++++++ tests/test_sync_activity_definitions.py | 95 ++++++++++++- ...WP-0032-glas-profile-execution-contract.md | 25 +++- 12 files changed, 455 insertions(+), 9 deletions(-) diff --git a/.env.example b/.env.example index e6914b6..36595a2 100644 --- a/.env.example +++ b/.env.example @@ -44,6 +44,11 @@ OPS_RUN_LEASE_SECONDS=900 OPS_RUN_MAX_ATTEMPTS=3 # Stuck open/claimed threshold for /ops/automations/status ops_runs.sla_hours OPS_RUN_SLA_HOURS=1 +# Migration gate for ACT-ADR-006. When true, definition sync rejects task- +# emitting declarations without a profile and emit_tasks refuses the whole +# batch before any DB or IssueSink side effect. Leave false until definitions +# have migrated one at a time. +ACTIVITY_CORE_REQUIRE_HARNESS_PROFILE=false # Worker credential for claim/complete/fail (X-Worker-Token or Bearer). ACTIVITY_CORE_WORKER_TOKEN= diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 91549a1..0a06c31 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -197,7 +197,7 @@ | task | ACTIVITY-WP-0031-T05 | wait | — | workplans/ACTIVITY-WP-0031-production-execution-reliability-cleanup.md | | task | ACTIVITY-WP-0032-T01 | done | — | workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md | | task | ACTIVITY-WP-0032-T02 | done | — | workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md | -| task | ACTIVITY-WP-0032-T03 | todo | — | workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md | +| task | ACTIVITY-WP-0032-T03 | done | — | workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md | | task | ACTIVITY-WP-0032-T04 | wait | — | workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md | | task | ACTIVITY-WP-0032-T05 | wait | — | workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md | | task | ACTIVITY-WP-0033-T01 | done | — | workplans/ACTIVITY-WP-0033-sbom-catchup-retry-boundary.md | diff --git a/docs/ops-run-queue.md b/docs/ops-run-queue.md index 54cf4b9..f05a040 100644 --- a/docs/ops-run-queue.md +++ b/docs/ops-run-queue.md @@ -162,6 +162,27 @@ A well-formed but *unknown* profile is still caught by the execution-side Glas resolver rather than at emission. That residual gap is accepted and recorded in ACT-ADR-006; closing it needs a scoped glas-harness API, not a local catalogue. +Task-emitting rules declare the selector and optional attribution refs on the +action; instructions use the same fields at instruction level: + +```yaml +action: + task_template: Run controlled maintenance + harness_profile_ref: harness.agent-dev@1.0.0 + approach_hint: legacy-definition-match + execution_refs: + correlation_id: context.request.correlation_id + goal_refs: [context.request.goal_ref] +``` + +File sync validates every declared profile structurally and rejects malformed +or unversioned refs. `emit_tasks` repeats that validation across the complete +batch before opening the database or IssueSink. A profile policy failure is a +non-retryable activity error; no earlier item in that batch is emitted. Only the +allowlisted attribution keys are carried to the queue. + `ACTIVITY_CORE_REQUIRE_HARNESS_PROFILE=true` makes a missing profile ref an -error. It stays off during coexistence while definitions adopt refs one at a -time; turn it on once no caller depends on `approach_hint` for routing. +error both during file sync and emission. Deterministic report-only +instructions are exempt because they create no execution request. The flag +stays off during coexistence while definitions adopt refs one at a time; turn it +on once no caller depends on `approach_hint` for routing. diff --git a/src/activity_core/activities.py b/src/activity_core/activities.py index 4d35fde..632f846 100644 --- a/src/activity_core/activities.py +++ b/src/activity_core/activities.py @@ -503,6 +503,9 @@ async def evaluate_instructions(payload: dict) -> dict: "model": result.model, "output_validated": result.output_validated, "review_required": result.review_required, + "approach_hint": instruction.approach_hint, + "harness_profile_ref": instruction.harness_profile_ref, + "execution_refs": instruction.execution_refs, }) return {"task_specs": task_specs, "reports": reports} @@ -538,6 +541,41 @@ async def emit_tasks(payload: dict) -> list[str]: activity_id = payload.get("activity_id", "") triggering_event_id = payload.get("triggering_event_id", "") + # Profile errors are policy failures, not best-effort queue failures. Check + # the whole batch before opening a DB transaction or touching IssueSink so a + # later malformed item cannot leave an earlier item partially emitted. + from activity_core.glas_profile import ( + ProfileRefError, + normalise_execution_refs, + resolve_execution_selector, + ) + + validated_specs: list[dict] = [] + for index, raw_spec in enumerate(task_specs_raw): + if not isinstance(raw_spec, dict): + raise ApplicationError( + f"task spec {index} is not a mapping", + non_retryable=True, + ) + spec_dict = dict(raw_spec) + try: + profile_ref, approach_hint = resolve_execution_selector( + spec_dict.get("harness_profile_ref"), + spec_dict.get("approach_hint"), + ) + except ProfileRefError as exc: + source = f"{spec_dict.get('source_type', 'rule')}:{spec_dict.get('source_id', '')}" + raise ApplicationError( + f"execution profile refused for {source}: {exc}", + non_retryable=True, + ) from exc + spec_dict["harness_profile_ref"] = profile_ref + spec_dict["approach_hint"] = approach_hint + spec_dict["execution_refs"] = normalise_execution_refs( + spec_dict.get("execution_refs") + ) + validated_specs.append(spec_dict) + sink = get_issue_sink() Session = _get_session_factory() @@ -545,7 +583,7 @@ async def emit_tasks(payload: dict) -> list[str]: errors: list[str] = [] async with Session() as session: async with session.begin(): - for spec_dict in task_specs_raw: + for spec_dict in validated_specs: spec = TaskSpec( title=spec_dict.get("title", ""), description=spec_dict.get("description", ""), diff --git a/src/activity_core/definition_parser.py b/src/activity_core/definition_parser.py index 086fc90..7941f71 100644 --- a/src/activity_core/definition_parser.py +++ b/src/activity_core/definition_parser.py @@ -17,6 +17,12 @@ from typing import Any import yaml +from activity_core.glas_profile import ( + ProfileRefError, + require_harness_profile, + validate_profile_ref, +) + class ParseError(Exception): """Raised when a definition file cannot be parsed.""" @@ -50,6 +56,72 @@ _FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL) _FENCED_BLOCK_RE = re.compile(r"^```(\w+)\s*\n(.*?)\n```", re.DOTALL | re.MULTILINE) +def _instruction_can_emit_tasks(instruction: dict[str, Any]) -> bool: + """Return false only for the explicit deterministic report-only shape.""" + model = str(instruction.get("model") or "").strip().lower() + return not ( + model in {"none", "deterministic", "unused"} + and bool(instruction.get("report_sinks")) + ) + + +def _validate_execution_declarations( + rules: list[dict[str, Any]], + instructions: list[dict[str, Any]], + file: Path, +) -> None: + """Validate declared profile structure without mirroring the Glas catalog.""" + profile_required = require_harness_profile() + declarations: list[tuple[str, dict[str, Any], bool]] = [] + + for rule in rules: + action = rule.get("action") + if not isinstance(action, dict): + raise ParseError( + file, + None, + f"rule {rule.get('id')!r} action must be a YAML mapping", + ) + declarations.append((f"rule {rule.get('id')!r}", action, True)) + + for instruction in instructions: + declarations.append( + ( + f"instruction {instruction.get('id')!r}", + instruction, + _instruction_can_emit_tasks(instruction), + ) + ) + + for location, declaration, can_emit_tasks in declarations: + raw_profile = declaration.get("harness_profile_ref") + if raw_profile is None: + if profile_required and can_emit_tasks: + raise ParseError( + file, + None, + f"{location} must declare harness_profile_ref while " + "ACTIVITY_CORE_REQUIRE_HARNESS_PROFILE is enabled", + ) + else: + try: + declaration["harness_profile_ref"] = validate_profile_ref(raw_profile) + except ProfileRefError as exc: + raise ParseError( + file, + None, + f"{location} has invalid harness_profile_ref: {exc}", + ) from exc + + execution_refs = declaration.get("execution_refs") + if execution_refs is not None and not isinstance(execution_refs, dict): + raise ParseError( + file, + None, + f"{location} execution_refs must be a YAML mapping", + ) + + def _scan_dirs() -> list[Path]: dirs: list[Path] = [] default_dir = Path("activity-definitions") @@ -164,6 +236,8 @@ def parse_file(path: Path) -> ActivityDefinitionDef: raise ParseError(path, None, "instruction block missing required field 'id'") instructions.append(block_data) + _validate_execution_declarations(rules, instructions, path) + return ActivityDefinitionDef( id=str(fm["id"]), name=str(fm["name"]), diff --git a/src/activity_core/models.py b/src/activity_core/models.py index ff25315..94038a3 100644 --- a/src/activity_core/models.py +++ b/src/activity_core/models.py @@ -99,6 +99,9 @@ class ActionDef(BaseModel): priority: str = Field(default="medium") labels: list[str] = Field(default_factory=list) due_in_days: int | None = Field(default=None) + approach_hint: str | None = Field(default=None) + harness_profile_ref: str | None = Field(default=None) + execution_refs: dict[str, Any] = Field(default_factory=dict) class RuleDef(BaseModel): @@ -136,6 +139,9 @@ class InstructionDef(BaseModel): output_schema: str = Field(description="Path to JSON Schema file for output validation.") review_required: bool = Field(default=False) report_sinks: list[dict[str, Any]] = Field(default_factory=list) + approach_hint: str | None = Field(default=None) + harness_profile_ref: str | None = Field(default=None) + execution_refs: dict[str, Any] = Field(default_factory=dict) # ── Context sources ─────────────────────────────────────────────────────────── diff --git a/src/activity_core/rules/actions.py b/src/activity_core/rules/actions.py index d1c8a5b..7292f14 100644 --- a/src/activity_core/rules/actions.py +++ b/src/activity_core/rules/actions.py @@ -80,10 +80,31 @@ def _task_spec_for_rule(rule: dict, event: Any, context: dict) -> dict: source_id=rule.get("id", ""), ) result = asdict(spec) + for key in ("approach_hint", "harness_profile_ref"): + if key in action: + result[key] = _string_or_none( + _render_value(action.get(key), event, context) + ) + if "execution_refs" in action: + result["execution_refs"] = _render_execution_refs( + action.get("execution_refs"), event, context + ) result["condition"] = rule.get("condition", "") return result +def _render_execution_refs(value: Any, event: Any, context: dict) -> dict[str, Any]: + if not isinstance(value, dict): + return {} + rendered: dict[str, Any] = {} + for key, raw in value.items(): + if isinstance(raw, list): + rendered[key] = [_render_value(item, event, context) for item in raw] + else: + rendered[key] = _render_value(raw, event, context) + return rendered + + def _render_labels(value: Any, event: Any, context: dict) -> list[str]: if not isinstance(value, list): return [] diff --git a/tests/rules/test_actions.py b/tests/rules/test_actions.py index 010e309..fea4bfb 100644 --- a/tests/rules/test_actions.py +++ b/tests/rules/test_actions.py @@ -54,6 +54,38 @@ def test_action_field_path_interpolation_resolves_context_value() -> None: ] +def test_action_carries_declared_profile_and_renders_attribution_refs() -> None: + rules = [ + { + "id": "profiled-task", + "condition": "", + "action": { + "task_template": "Run controlled task", + "target_repo": "activity-core", + "harness_profile_ref": "harness.agent-dev@1.0.0", + "approach_hint": "legacy-coding", + "execution_refs": { + "correlation_id": "context.request.correlation_id", + "goal_refs": ["context.request.goal_ref"], + }, + }, + } + ] + + specs = expand_rule_actions( + rules, + _Event(), + {"request": {"correlation_id": "corr-7", "goal_ref": "goal:7@1"}}, + ) + + assert specs[0]["harness_profile_ref"] == "harness.agent-dev@1.0.0" + assert specs[0]["approach_hint"] == "legacy-coding" + assert specs[0]["execution_refs"] == { + "correlation_id": "corr-7", + "goal_refs": ["goal:7@1"], + } + + def test_for_each_binds_each_list_item_before_condition_and_action_rendering() -> None: rules = [ { diff --git a/tests/test_glas_profile_selection.py b/tests/test_glas_profile_selection.py index 6128433..3eb1b1c 100644 --- a/tests/test_glas_profile_selection.py +++ b/tests/test_glas_profile_selection.py @@ -100,6 +100,14 @@ class TestNoFallbackToApproachHint: assert profile == "harness.agent-dev@2.1.0" + def test_unknown_but_well_formed_profile_is_left_for_glas(self) -> None: + profile, _ = resolve_execution_selector( + "harness.not-in-any-local-catalog@9.9.9", + None, + ) + + assert profile == "harness.not-in-any-local-catalog@9.9.9" + def test_blank_hint_is_normalised_away(self) -> None: assert resolve_execution_selector(None, " ") == (None, None) diff --git a/tests/test_issue_sink.py b/tests/test_issue_sink.py index 2383881..202fa30 100644 --- a/tests/test_issue_sink.py +++ b/tests/test_issue_sink.py @@ -175,3 +175,134 @@ async def test_emit_tasks_raises_when_sink_fails(monkeypatch) -> None: } ], }) + + +@pytest.mark.asyncio +async def test_emit_tasks_refuses_malformed_profile_before_any_side_effect( + monkeypatch, +) -> None: + from temporalio.exceptions import ApplicationError + + monkeypatch.setattr( + activities, + "get_issue_sink", + lambda: pytest.fail("IssueSink must not be opened after profile refusal"), + ) + monkeypatch.setattr( + activities, + "_get_session_factory", + lambda: pytest.fail("DB must not be opened after profile refusal"), + ) + + with pytest.raises(ApplicationError, match="must pin a version") as exc_info: + await activities.emit_tasks( + { + "activity_id": "00000000-0000-0000-0000-000000000001", + "triggering_event_id": "scheduled", + "task_specs": [ + { + "title": "Must not emit", + "source_type": "rule", + "source_id": "malformed-profile", + "harness_profile_ref": "harness.agent-dev", + "approach_hint": "legacy-must-not-rescue", + } + ], + } + ) + + assert exc_info.value.non_retryable is True + + +@pytest.mark.asyncio +async def test_emit_tasks_refuses_absent_profile_in_strict_mode(monkeypatch) -> None: + from temporalio.exceptions import ApplicationError + + monkeypatch.setenv("ACTIVITY_CORE_REQUIRE_HARNESS_PROFILE", "true") + monkeypatch.setattr( + activities, + "get_issue_sink", + lambda: pytest.fail("IssueSink must not be opened after profile refusal"), + ) + + with pytest.raises(ApplicationError, match="cannot substitute"): + await activities.emit_tasks( + { + "activity_id": "00000000-0000-0000-0000-000000000001", + "triggering_event_id": "scheduled", + "task_specs": [ + { + "title": "Must not emit", + "source_type": "rule", + "source_id": "missing-profile", + "approach_hint": "legacy-must-not-rescue", + } + ], + } + ) + + +@pytest.mark.asyncio +async def test_emit_tasks_passes_unknown_versioned_profile_to_queue(monkeypatch) -> None: + captured: dict[str, Any] = {} + + class SuccessfulSink: + def emit(self, task_spec: TaskSpec) -> TaskRef: + return TaskRef(external_id="state-hub:progress-1", backend="state-hub") + + class FakeTransaction: + async def __aenter__(self) -> None: + return None + + async def __aexit__(self, *exc_info: object) -> bool: + return False + + class FakeSession: + def begin(self) -> FakeTransaction: + return FakeTransaction() + + async def __aenter__(self) -> "FakeSession": + return self + + async def __aexit__(self, *exc_info: object) -> bool: + return False + + def add(self, row: object) -> None: + return None + + class FakeSessionFactory: + def __call__(self) -> FakeSession: + return FakeSession() + + async def fake_create_ops_run(session, spec, **kwargs): + captured.update(kwargs) + return None + + monkeypatch.setattr(activities, "get_issue_sink", lambda: SuccessfulSink()) + monkeypatch.setattr(activities, "_get_session_factory", lambda: FakeSessionFactory()) + monkeypatch.setattr(activities, "create_ops_run_from_spec", fake_create_ops_run) + + refs = await activities.emit_tasks( + { + "activity_id": "00000000-0000-0000-0000-000000000001", + "triggering_event_id": "event-1", + "task_specs": [ + { + "title": "Let Glas resolve it", + "source_type": "rule", + "source_id": "profiled", + "harness_profile_ref": "harness.unknown-locally@9.9.9", + "approach_hint": "legacy-only", + "execution_refs": { + "correlation_id": "corr-9", + "api_key": "must-be-dropped", + }, + } + ], + } + ) + + assert refs == ["state-hub:progress-1"] + assert captured["harness_profile_ref"] == "harness.unknown-locally@9.9.9" + assert captured["approach_hint"] == "legacy-only" + assert captured["execution_refs"] == {"correlation_id": "corr-9"} diff --git a/tests/test_sync_activity_definitions.py b/tests/test_sync_activity_definitions.py index f34a7e3..4384901 100644 --- a/tests/test_sync_activity_definitions.py +++ b/tests/test_sync_activity_definitions.py @@ -1,6 +1,8 @@ import uuid -from activity_core.definition_parser import scan_and_parse +import pytest + +from activity_core.definition_parser import ParseError, parse_file, scan_and_parse from activity_core.models import ActivityDefinition from activity_core.sync_activity_definitions import _definition_uuid @@ -95,3 +97,94 @@ context_sources: assert definition.context_sources[0]["params"]["evidence_sinks"][0]["type"] == ( "state-hub-progress" ) + + +def _write_profiled_definition(tmp_path, action_lines: str) -> None: + (tmp_path / "profiled.md").write_text( + f"""--- +id: profiled +name: Profiled +enabled: true +trigger: + type: cron + cron_expression: "0 9 * * *" +--- + +```rule +id: emit +condition: "" +action: + task_template: Run it +{action_lines} +``` +""", + encoding="utf-8", + ) + + +def test_definition_parse_normalises_a_versioned_profile(tmp_path) -> None: + _write_profiled_definition( + tmp_path, + " harness_profile_ref: ' harness.agent-dev@1.0.0 '\n", + ) + + definition = parse_file(tmp_path / "profiled.md") + + assert definition.rules[0]["action"]["harness_profile_ref"] == ( + "harness.agent-dev@1.0.0" + ) + + +def test_definition_parse_rejects_a_malformed_profile(tmp_path) -> None: + _write_profiled_definition( + tmp_path, + " harness_profile_ref: harness.agent-dev\n", + ) + + with pytest.raises(ParseError, match="must pin a version"): + parse_file(tmp_path / "profiled.md") + + +def test_definition_parse_requires_rule_profile_in_strict_mode( + tmp_path, + monkeypatch, +) -> None: + _write_profiled_definition(tmp_path, "") + monkeypatch.setenv("ACTIVITY_CORE_REQUIRE_HARNESS_PROFILE", "true") + + with pytest.raises(ParseError, match="must declare harness_profile_ref"): + parse_file(tmp_path / "profiled.md") + + +def test_strict_mode_does_not_require_profile_for_deterministic_report( + tmp_path, + monkeypatch, +) -> None: + path = tmp_path / "report.md" + path.write_text( + """--- +id: report +name: Report only +enabled: true +trigger: + type: cron + cron_expression: "0 9 * * *" +--- + +```instruction +id: report-only +trusted_fields: [] +model: deterministic +prompt: Report +output_schema: "" +report_sinks: + - type: state-hub-progress +``` +""", + encoding="utf-8", + ) + monkeypatch.setenv("ACTIVITY_CORE_REQUIRE_HARNESS_PROFILE", "true") + + definition = parse_file(path) + + assert definition.instructions[0]["id"] == "report-only" diff --git a/workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md b/workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md index ac0ecd6..d34195c 100644 --- a/workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md +++ b/workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md @@ -154,7 +154,7 @@ silent fallback is exactly the failure this workplan removes. ```task id: ACTIVITY-WP-0032-T03 -status: todo +status: done priority: medium state_hub_task_id: "50225bac-1798-5cf5-91ad-d4c662de7fc7" ``` @@ -177,6 +177,23 @@ Glas refuses before sandbox creation. If emit-time remote validation becomes necessary, raise it as a capability request against glas-harness rather than solving it locally. +Done 2026-08-22. Definition parsing now validates every declared rule or +instruction profile structurally and normalizes its version-pinned ref before +DB sync. With `ACTIVITY_CORE_REQUIRE_HARNESS_PROFILE=true`, sync refuses +task-emitting declarations that omit a profile; deterministic report-only +instructions remain exempt. Rule and instruction expansion carry the declared +profile, legacy hint, and allowlisted attribution refs into the queued request. + +Emission now preflights the complete task batch before opening either the DB or +IssueSink. A missing profile in strict mode or any malformed profile becomes a +non-retryable activity error, so a later invalid item cannot leave earlier +items partially emitted and `approach_hint` cannot rescue it. Tests also prove +that an unknown but structurally valid versioned ref is preserved for the +authoritative execution-side Glas resolver; no local profile catalogue was +introduced. Verification: 59 focused tests and 426 repository tests passed; +the one live NATS-to-Temporal bridge test was deselected because it requires +the local integration stack. Python compilation and `git diff --check` passed. + ## Record normalized execution evidence ```task @@ -211,10 +228,10 @@ provider-credential failures and would mask the result. - [x] Invocation shape decided and recorded as an ADR, with the boundary stated (ACT-ADR-006; pull queue preserved) -- [ ] ops_run carries an approved `harness_profile_ref`; no definition names a +- [x] ops_run carries an approved `harness_profile_ref`; no definition names a concrete rein -- [ ] Malformed/absent profile refs are refused at emission; unknown-but-well-formed +- [x] Malformed/absent profile refs are refused at emission; unknown-but-well-formed refs are refused by Glas before sandbox creation, never at claim -- [ ] `approach_hint` cannot override or substitute for a profile ref, proven by test +- [x] `approach_hint` cannot override or substitute for a profile ref, proven by test - [ ] Normalized Glas evidence is visible in production status - [ ] One definition proven on railiance01 end to end