From 26934e25b9fb0f18ed24d4d1883eb0fc6210b12f Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 23 Aug 2026 12:31:13 +0200 Subject: [PATCH] Enforce bounded operation guardrails Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028de-e2c8-7732-8521-46a7fc5db82f --- INTENT.md | 53 ++- README.md | 29 +- SCOPE.md | 25 +- .../daily-cnpg-option-a-backup.md | 1 + activity-definitions/daily-sbom-catchup.md | 3 +- .../weekly-forgejo-package-prune.md | 1 + activity-definitions/weekly-sbom-staleness.md | 2 +- docs/adr/adr-001-event-bridge-architecture.md | 13 +- docs/adr/adr-002-definition-format.md | 2 +- docs/adr/adr-003-rule-instruction-model.md | 39 +- docs/adr/adr-004-producer-trust-boundary.md | 2 +- .../adr-005-ops-runs-vs-dev-work-records.md | 5 + docs/adr/adr-006-glas-profile-execution.md | 5 + docs/adr/adr-007-bounded-operations.md | 128 ++++++ docs/conventions.md | 13 +- docs/execution-queue-boundary.md | 4 + docs/idempotency.md | 19 +- docs/runbook.md | 12 +- docs/task-emission-consumer-contract.md | 4 +- e2e/tests/test_full_flow.py | 16 - ...-bounded-operation-compatibility-report.md | 43 ++ k8s/railiance/20-runtime.yaml | 5 +- .../0009_drop_legacy_task_instances.py | 57 +++ schemas/activity-definition.json | 399 ++++++++++++++++-- schemas/event-envelope.json | 52 +-- src/activity_core/activities.py | 124 ++++-- src/activity_core/api.py | 36 +- src/activity_core/audit_projection.py | 68 +++ src/activity_core/bounded_operations.py | 292 +++++++++++++ .../context_resolvers/reuse_surface.py | 12 +- src/activity_core/definition_parser.py | 28 ++ src/activity_core/models.py | 34 +- src/activity_core/ops_evidence_sinks.py | 9 +- src/activity_core/orm.py | 22 +- src/activity_core/report_sinks.py | 11 +- src/activity_core/rules/executor.py | 61 +-- src/activity_core/run_artifacts.py | 11 +- src/activity_core/worker.py | 50 +-- src/activity_core/workflows.py | 113 ++--- tests/rules/test_executor.py | 42 +- tests/test_api_admission.py | 28 ++ tests/test_audit_projection.py | 40 ++ tests/test_bounded_operations.py | 167 ++++++++ tests/test_forgejo_prune_context_resolver.py | 61 ++- tests/test_instruction_evaluation.py | 19 +- tests/test_ops_evidence_sinks.py | 48 +++ tests/test_report_sinks.py | 10 +- tests/test_resolve_context_binding.py | 34 ++ tests/test_run_artifacts.py | 12 + tests/test_sbom_nexus_context_resolver.py | 2 +- ...VITY-WP-0035-intent-boundary-guardrails.md | 49 ++- 51 files changed, 1843 insertions(+), 472 deletions(-) create mode 100644 docs/adr/adr-007-bounded-operations.md create mode 100644 history/2026-08-23-bounded-operation-compatibility-report.md create mode 100644 migrations/versions/0009_drop_legacy_task_instances.py create mode 100644 src/activity_core/audit_projection.py create mode 100644 src/activity_core/bounded_operations.py create mode 100644 tests/test_api_admission.py create mode 100644 tests/test_audit_projection.py create mode 100644 tests/test_bounded_operations.py diff --git a/INTENT.md b/INTENT.md index d80c2c5..faa2079 100644 --- a/INTENT.md +++ b/INTENT.md @@ -1,7 +1,7 @@ --- domain: capabilities repo: activity-core -updated: "2026-08-18" +updated: "2026-08-23" --- # INTENT @@ -35,8 +35,23 @@ activity-core answers three questions and only three: 2. **What** — given current org context, what work must be created? 3. **Where** — which repo, service, or agent should each work item land? -It does not execute the work. It does not track task lifecycle. It does not -manage projects or campaigns. Those belong to other systems. +It does not execute general domain work. It does not track workplan, tracker, +or human task lifecycle. It does not manage projects or campaigns. Those belong +to other systems. + +Two narrow runtime mechanisms support the three answers without adding a +fourth responsibility: + +- an `ops_run` is a durable delivery record for one automation fire. Its + open/claim/lease/outcome state is runtime operations data, not a Task, + Commitment, assignment, or project record; and +- a code-registered **bounded operation** may perform the declared maintenance + outcome directly only under ACT-ADR-007: fixed/bounded targets, explicit + mutation intent, operation-specific idempotency/retry rules, routed + credentials, and mandatory non-secret evidence. + +Neither mechanism is permission to accept arbitrary commands, execute agent +loops, edit repositories, or grow task/project semantics here. This constraint is intentional and load-bearing. An orchestrator that also stores task state, manages project phases, or executes work becomes a God object @@ -71,6 +86,14 @@ It is an event loop governed by declarative rules and LLM instructions: records the spawn event as an audit trail — not as the authoritative task record. Downstream **executors** (per-repo workers, agent-harness) perform the work. +- **Runtime delivery**: internal automation work may also be represented by a + claimable `ops_run`. activity-core owns its delivery lease and normalized + outcome evidence, while Glas/reins or per-repo workers perform the domain + work. +- **Bounded operations**: a small code-owned registry permits maintenance whose + whole purpose is a fixed, reviewable operation such as controlled SBOM ingest, + package retention, or backup. Context resolution remains read-only; the + mutation runs in an explicit workflow stage governed by ACT-ADR-007. --- @@ -89,7 +112,8 @@ It is an event loop governed by declarative rules and LLM instructions: | Mixed coordination read model (until retirement) | state-hub (STATE-WP-0079 strangler; not a permanent peer) | | Project and initiative management | `prj-*` repos + GOAL.md | | Repository capability profiling | repo-scoping | -| Execution of domain work | per-repo workers / rein-aharness claiming ops_runs | +| General execution of domain work | Glas/reins and per-repo workers claiming `ops_runs` | +| Bounded maintenance operation implementation | activity-core only for ACT-ADR-007 registered operations; otherwise platform owner | | Event broker infrastructure | NATS (org infrastructure) | activity-core does not compete with those owners. It **schedules and evaluates**; @@ -115,9 +139,11 @@ When activity-core is in place, Bernd can: domain standards" — and have an LLM agent make that judgement reproducibly. - Set up a one-off reminder — "on 2026-09-01, create a review task for the Q3 architecture retrospective" — without managing a separate reminder system. -- Observe a complete audit trail of every activation: what triggered it, what - rules matched, what tasks were created, and (for instructions) what prompt - and model produced the output. +- Observe a bounded, non-secret audit trail of every activation: what triggered + it, what definition version and context were used, what rules matched, what + outputs were emitted, and (for instructions) the model, validation result, + and SHA-256 hash of the rendered prompt. The hash is an integrity reference, + not storage or guaranteed reconstruction of the raw prompt. The Coulomb org gains **structured, auditable automation** that scales with the number of repos and domains without scaling the coordination burden on Bernd. @@ -136,10 +162,15 @@ instructions when the condition is fully expressible. Instructions are reserved for genuine judgement cases. This keeps most automation fast, cheap, testable, and auditable. -**No task state ownership.** activity-core holds a spawn audit trail, not task -state. The moment it starts tracking whether tasks are complete, blocked, or -re-assigned, it has become a task database — that belongs to issue-core or -fleet work-record systems, not here. +**No work-item state ownership.** activity-core holds spawn audit plus runtime +delivery state for `ops_runs`, not Task/Commitment state. The moment it starts +tracking whether workplan or tracker tasks are assigned, blocked, completed, +dependent, or re-assigned, it has become a task database — that belongs to +issue-core or fleet work-record systems, not here. + +**Explicit bounded operations only.** Context adapters read. A mutation may run +only through the ACT-ADR-007 registry and workflow phase; unknown operation +names and incomplete safety declarations fail during definition admission. **Safe sinks for internal findings.** Automated internal coordination must not default to silent Forgejo/Gitea issues. Prefer State Hub progress / work-record diff --git a/README.md b/README.md index fcd7b8f..bf60f36 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,28 @@ -# repo-seed +# activity-core -A git repository template to bootstrap coulomb projects from. \ No newline at end of file +activity-core is Coulomb's durable automation factory. It answers **when** an +automation fires (Temporal Schedules, NATS, webhooks, or manual triggers), +**what** should be emitted (deterministic rules or optional validated LLM +instructions), and **where** the result goes (task, report, evidence, or +runtime-delivery sinks). + +It is not a project tracker, a general shell runner, or a task executor. A +small code-owned registry permits three bounded platform operations: controlled +SBOM ingest, Forgejo package pruning, and CNPG Option A backup. + +## Start here + +- [INTENT.md](INTENT.md) — product purpose and hard boundaries +- [SCOPE.md](SCOPE.md) — implemented capability and current gaps +- [docs/runbook.md](docs/runbook.md) — local and production operations +- [docs/api.md](docs/api.md) — API surface +- [docs/adr/adr-007-bounded-operations.md](docs/adr/adr-007-bounded-operations.md) + — admitted mutation contract + +Definitions are authored in `activity-definitions/*.md` and synchronized into +Postgres and Temporal. The `/activity-definitions/` REST endpoints provide +basic row administration only; they do not round-trip the full markdown +rules/instructions contract. + +For local setup and verification, follow the runbook. The installed `activity` +CLI is the operator review surface; run `uv run activity --help` for commands. diff --git a/SCOPE.md b/SCOPE.md index 87382c2..a0f1585 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -132,11 +132,11 @@ The two evaluation modes: not task lifecycle. - **Coding assistant schedulers** as production authority. -**Boundary note (side-effect resolvers):** A small set of shell/context -resolvers intentionally invoke platform tools (e.g. Forgejo package prune). -That is allowed only when the side-effect is the *declared purpose* of the -ActivityDefinition, is credential-routed, evidence-posted, and not general -task execution. It must not expand into an unbounded ops executor. +**Boundary note (ACT-ADR-007):** Context resolution is read-only. A small +code-owned registry permits explicit bounded operations (currently controlled +SBOM ingest, Forgejo package prune, and CNPG backup) in a separate workflow +phase. Unknown operations or incomplete mutation/evidence declarations are not +an extension mechanism and must fail definition admission. --- @@ -230,22 +230,18 @@ The detailed review is preserved in | Markdown definitions | Met — `activity-definitions/`, event-types, external ConfigMaps | | Rules before instructions | Met — deterministic rule/report paths dominate; LLM is optional | | Durable recurring automation | Met — Temporal schedules plus deterministic status evidence | -| Audit trail | Partly met — activation, context, prompt hash/model, emission, queue, and normalized execution evidence exist; raw prompt text is intentionally absent | -| Does not own task lifecycle | Met for workplan/tracker tasks; qualified by an in-repo machine `ops_run` claim/lease/outcome lifecycle | +| Audit trail | Met as a bounded, non-secret audit projection — activation, definition/version, context projection, prompt hash/model, validation, emission, queue, and normalized execution evidence; raw prompts/provider payloads are intentionally absent | +| Does not own task lifecycle | Met — `ops_run` claim/lease/outcome state is runtime delivery state, not work-item lifecycle | | Does not own project phases | Met | ### Gaps and tensions (INTENT ↔ practice) | Gap | Severity | Notes | | --- | --- | --- | -| **G1. Operational state boundary** | High (clarity) | `ops_runs` owns open/claimed/terminal machine-run state. This is not work-item lifecycle, but INTENT's absolute wording does not state the distinction. | -| **G2. Bounded side effects** | Medium (boundary) | SBOM ingest, package prune, and CNPG backup execute declared operations inside the orchestration workflow; INTENT says activity-core does not execute work. | | **G3. Live LLM execution** | High (operational, external) | The implementation is present, but production provider requests fail with a sanitized upstream 401 until the account owner replaces the key. | | **G4. Profiled execution proof** | Medium (operational, external) | Profile selection, failure evidence, and teardown are proven; a successful commit waits on `GLAS-IN-0002`. | | **G5. State Hub retirement** | Medium (integration) | hub-core adapters exist, but production task/report progress still defaults to the State Hub compatibility path. | -| **G6. Review routing** | Medium | `review_required` remains evidence metadata; there is no downstream pending-review queue. | -| **G7. Audit wording** | Medium (assurance) | The repo records a prompt hash, model, definition version, and context snapshot, not the literal rendered prompt promised by INTENT. | -| **G8. Legacy surfaces** | Low | `TaskExecutorWorkflow`/`task_instances` remain disabled compatibility residue, and the basic REST CRUD schema cannot author full rules/instructions. | +| **G6. Definition REST parity** | Low | Basic REST row administration intentionally cannot author or round-trip full rules/instructions; markdown source sync is authoritative. | ### Drift risks @@ -257,8 +253,6 @@ The detailed review is preserved in fresh live-image lists. 4. **Queue/task conflation** — treating an `ops_run` as the authoritative human task record rather than a delivery instance. -5. **Stub workflow attraction** — `TaskExecutorWorkflow` looks like an - execution home despite being disabled by default. --- @@ -272,7 +266,7 @@ The detailed review is preserved in [IssueSink: rest | state-hub | null] → issue-core or hub-core progress (State Hub until cutover) [report/evidence sinks] → hub-core / working memory [ops_run claim queue + Glas profile] → rein-aharness / other approved reins -[bounded shell side-effects] → platform tools (e.g. package prune) +[bounded operation stage] → code-registered platform tools ``` - **Upstream**: NATS, Temporal, PostgreSQL, repo-manager / hub-core (State Hub @@ -321,6 +315,7 @@ The detailed review is preserved in - `docs/adr/adr-004-producer-trust-boundary.md` - `docs/adr/adr-005-ops-runs-vs-dev-work-records.md` - `docs/adr/adr-006-glas-profile-execution.md` +- `docs/adr/adr-007-bounded-operations.md` --- diff --git a/activity-definitions/daily-cnpg-option-a-backup.md b/activity-definitions/daily-cnpg-option-a-backup.md index 3cc0da0..4d5e69c 100644 --- a/activity-definitions/daily-cnpg-option-a-backup.md +++ b/activity-definitions/daily-cnpg-option-a-backup.md @@ -13,6 +13,7 @@ trigger: context_sources: - type: shell query: cnpg_option_a_backup + operation: cnpg_option_a_backup required: true params: backup_script: /opt/railiance-platform/tools/cmd/cnpg-option-a-backup diff --git a/activity-definitions/daily-sbom-catchup.md b/activity-definitions/daily-sbom-catchup.md index fd0f460..45bec0b 100644 --- a/activity-definitions/daily-sbom-catchup.md +++ b/activity-definitions/daily-sbom-catchup.md @@ -13,6 +13,7 @@ trigger: context_sources: - type: sbom-nexus query: catch_up + operation: sbom_nexus_ingest required: true params: limit: 3 # catch_up_limit — operator knob, not a nexus constant @@ -67,7 +68,7 @@ max_tokens: 1 prompt: | Deterministic SBOM catch-up report from context.catchup (no LLM). output_schema: "" -review_required: false +review_advisory: false report_sinks: - type: state-hub-progress event_type: sbom_catchup diff --git a/activity-definitions/weekly-forgejo-package-prune.md b/activity-definitions/weekly-forgejo-package-prune.md index 5022fd8..02f707a 100644 --- a/activity-definitions/weekly-forgejo-package-prune.md +++ b/activity-definitions/weekly-forgejo-package-prune.md @@ -13,6 +13,7 @@ trigger: context_sources: - type: shell query: forgejo_package_prune + operation: forgejo_package_prune required: true params: prune_script: /opt/railiance-platform/tools/cmd/forgejo-package-prune diff --git a/activity-definitions/weekly-sbom-staleness.md b/activity-definitions/weekly-sbom-staleness.md index 4a9869a..07f1691 100644 --- a/activity-definitions/weekly-sbom-staleness.md +++ b/activity-definitions/weekly-sbom-staleness.md @@ -49,7 +49,7 @@ max_tokens: 1 prompt: | Deterministic SBOM staleness report from context.repos (no LLM). output_schema: "" -review_required: false +review_advisory: false report_sinks: - type: state-hub-progress event_type: sbom_staleness diff --git a/docs/adr/adr-001-event-bridge-architecture.md b/docs/adr/adr-001-event-bridge-architecture.md index 743200e..edd8317 100644 --- a/docs/adr/adr-001-event-bridge-architecture.md +++ b/docs/adr/adr-001-event-bridge-architecture.md @@ -143,14 +143,11 @@ Three trigger types are supported: ### Immediate - activity-core's `INTENT.md` and `SCOPE.md` are rewritten to reflect this architecture. -- The `task_instances` Postgres table is reclassified as a **spawn audit trail** — - it records the act of spawning (what was created, when, which issue-core reference) - but is not the authoritative task record. Authoritative lifecycle state lives in - issue-core. -- A task emission adapter interface (`src/activity_core/issue_sink.py`) replaces any - direct Postgres writes to `task_instances` with calls through the adapter. -- The `TaskExecutorWorkflow` stub from WP-0001 is replaced with the actual adapter - call in WP-0003. +- `task_spawn_log` is the local **spawn audit trail**; authoritative work-item + lifecycle state lives downstream. +- A task emission adapter (`src/activity_core/issue_sink.py`) owns downstream + creation. The unused `TaskExecutorWorkflow` and `task_instances` compatibility + surface was retired by ACTIVITY-WP-0035. ### Medium term diff --git a/docs/adr/adr-002-definition-format.md b/docs/adr/adr-002-definition-format.md index 318e45b..d17dd64 100644 --- a/docs/adr/adr-002-definition-format.md +++ b/docs/adr/adr-002-definition-format.md @@ -227,7 +227,7 @@ trusted_fields: - event.attributes.domain - event.attributes.tags model: claude-sonnet-4-6 -review_required: false +review_advisory: false prompt: | A new repository has been registered in the Coulomb organization. diff --git a/docs/adr/adr-003-rule-instruction-model.md b/docs/adr/adr-003-rule-instruction-model.md index b2c6db3..6db1197 100644 --- a/docs/adr/adr-003-rule-instruction-model.md +++ b/docs/adr/adr-003-rule-instruction-model.md @@ -185,7 +185,7 @@ trusted_fields: # REQUIRED — explicit allowlist of payload - event.attributes.domain - event.attributes.tags model: claude-sonnet-4-6 -review_required: false # true | false — curator gate for output +review_advisory: false # true | false — advisory evidence, not a gate prompt: | {prompt template — only trusted_fields may be interpolated} output_schema: {path to JSON schema file} @@ -218,23 +218,19 @@ created from unvalidated output**. Structured output mode (tool_use / JSON mode) is used where the model supports it. The output schema must define `List[TaskSpec]` or a compatible envelope. -#### `review_required: true` +#### `review_advisory: true` -When set today, the instruction's task/report output is marked with -`review_required=true` in activity-core audit metadata. For report-producing -instructions, this flag is also persisted in configured report sinks so an -operator can distinguish validated-but-review-worthy output from routine -output. +ACTIVITY-WP-0035 selected advisory-only semantics because no downstream owner +currently exposes an acknowledged proposal/decision/release contract. The +instruction's task/report evidence is marked `review_advisory=true` and +`review_gate_applied=false` so an operator can distinguish review-worthy output +without interpreting it as held for approval. -activity-core does **not** currently route proposed tasks to a pending review -queue. That queue must be owned by issue-core, because issue-core owns task -lifecycle state. Until issue-core exposes a review contract, `review_required` -is metadata only; it must not be treated as evidence that live task creation was -held for approval. - -Future issue-core review integration may use the same field, but that change -must update the issue sink contract and tests before any ActivityDefinition -relies on queue routing. +The legacy input name `review_required` is accepted during migration and +normalized to `review_advisory`; new definitions must not use it. activity-core +does **not** route proposals to a pending-review queue and does not own review +lifecycle state. A future hold/release design requires a named downstream owner, +an idempotent release reference, an ADR update, and fail-closed emission tests. #### Evaluation semantics @@ -259,11 +255,20 @@ Every task emission records: | `prompt_hash` | — | SHA-256 of rendered prompt | | `model` | — | model ID used | | `output_validated` | — | `true` / `false` | -| `review_required` | — | `true` / `false` | +| `review_advisory` | — | `true` / `false`; no gate applied | The audit trail is written to the `task_spawn_log` table in activity-core's database and referenced from the task record in issue-core. +The rendered prompt and provider response are deliberately not persisted. The +prompt hash proves equality when an authorized operator can reconstruct the +same input, but the audit contract does not promise reconstruction after source +definitions or upstream event retention have changed. `activity_runs` retains +the definition version and bounded context snapshot; reports may retain +allowlisted route/usage metadata. Prompts, messages, tool output, credential +fields, and provider blobs are excluded from run, progress, and public API +evidence. + ### Testing strategy **Rules**: every rule can and should be unit-tested with fixture event payloads. diff --git a/docs/adr/adr-004-producer-trust-boundary.md b/docs/adr/adr-004-producer-trust-boundary.md index 5ccbfb7..5dd3269 100644 --- a/docs/adr/adr-004-producer-trust-boundary.md +++ b/docs/adr/adr-004-producer-trust-boundary.md @@ -103,7 +103,7 @@ Implemented in `src/activity_core/rules/executor.py`: pretty-printed and NDJSON output, attempts a best-effort `_try_repair` on a truncated tail, validates each recovered object against the item schema, and keeps the valid ones. Survivors are emitted with `output_validated=true`, - `partial=true`, and `review_required=true`. + `partial=true`, and `review_advisory=true` (`review_gate_applied=false`). - **Producer guardrails (`_partition_items`, applied on both the recovery and the happy path).** Per recommendation: structural type → schema → structural caps (`_MAX_DEPTH`, `_MAX_STRING_LEN`) → reference allow-list → count cap (top-N by diff --git a/docs/adr/adr-005-ops-runs-vs-dev-work-records.md b/docs/adr/adr-005-ops-runs-vs-dev-work-records.md index 4de7aeb..92b59cb 100644 --- a/docs/adr/adr-005-ops-runs-vs-dev-work-records.md +++ b/docs/adr/adr-005-ops-runs-vs-dev-work-records.md @@ -66,6 +66,11 @@ Canon already allows a DB-only exception for “runtime operations data (logs, metrics, run histories, token events)” (`work-record-types_v0.1.md`). This ADR names that exception for **ops runs**. +This is also the precise meaning of INTENT's no-task-lifecycle boundary: +`ops_run` claim/lease/outcome is durable runtime delivery state, not work-item +state. It may not grow assignment, commitments, dependencies, project phases, +or manually managed task status. + ## Decision ### 1. Two planes for “work,” one vocabulary for people diff --git a/docs/adr/adr-006-glas-profile-execution.md b/docs/adr/adr-006-glas-profile-execution.md index e3cef9a..0f5c8af 100644 --- a/docs/adr/adr-006-glas-profile-execution.md +++ b/docs/adr/adr-006-glas-profile-execution.md @@ -126,6 +126,11 @@ The attribution references we carry (`assignment_ref`, `role_ref`, `duty_ref`, here. Their vocabulary belongs to `info-tech-canon`; activity-core must not invent org roles (ACTIVITY-WP-0029 responsibility map). +ACT-ADR-007's bounded-operation exception does not alter this decision. A +code-registered fixed maintenance operation is not a Glas/rein execution +constellation, and the exception cannot be used to run an agent loop or execute +an `ops_run` inside activity-core. + ## Consequences - `ops_runs` grows `harness_profile_ref` and the attribution refs; the emission diff --git a/docs/adr/adr-007-bounded-operations.md b/docs/adr/adr-007-bounded-operations.md new file mode 100644 index 0000000..93a45f3 --- /dev/null +++ b/docs/adr/adr-007-bounded-operations.md @@ -0,0 +1,128 @@ +--- +id: ACT-ADR-007 +type: architecture-decision-record +title: "Code-registered bounded operations are the only local mutation exception" +status: accepted +state_hub_decision_id: "2b1f0c01-1d6f-4fdc-b3e7-537ce0a0a1f8" +owner: activity-core +revision: "accepted-1" +last_reviewed: "2026-08-23" +review_interval: 6m +decided_by: Bernd Worsch +date: "2026-08-23" +scope: repo +affects: + - activity-core + - sbom-nexus + - railiance-platform +tags: + - architecture + - activity-core + - bounded-operation + - execution-boundary +--- + +# ACT-ADR-007: Code-registered Bounded Operations + +## Status + +**Accepted** (2026-08-23) for ACTIVITY-WP-0035. + +## Context + +The Event Bridge principle says activity-core answers when, what, and where and +does not execute domain work. Production nevertheless contains three useful +operations whose complete outcome is a small scheduled maintenance action: + +- ingest at most three Repo Manager-selected immutable sources into SBOM Nexus; +- prune Forgejo package versions under live-image protection; and +- invoke the fixed CNPG Option A backup tool for an explicit target list. + +Historically the latter two ran through a generic `shell` context resolver. +That made a mutating subprocess look like a read and left extension policy to +convention. Removing the operations would recreate bespoke cron; accepting +arbitrary shell would make activity-core a general executor. + +## Decision + +### 1. Bounded operations are a narrow implementation exception + +The governing when/what/where responsibility does not gain a general “how.” A +bounded operation is allowed only when the operation itself is the declared +automation outcome and all admission requirements below are code-reviewable. + +The initial allowlist is exactly: + +1. `sbom_nexus_ingest` +2. `forgejo_package_prune` +3. `cnpg_option_a_backup` + +Adding an operation requires updating the code-owned registry, this ADR (or a +successor), tests, credential route, and evidence contract. A string in a +definition cannot register an operation. + +### 2. Admission is fail closed during file sync + +Each registry entry declares: + +- source type and query; +- whether mutation intent is `apply` or `dry_run` and that it is explicit; +- fixed or maximum target bounds; +- canonical implementation and allowed configuration; +- idempotency and Temporal retry semantics; +- maximum execution timeout; +- credential owner/route; and +- mandatory non-secret evidence mode. + +Unknown `shell` queries are refused unless separately registered as read-only. +Known operation queries with missing, malformed, or over-limit safety fields +are refused before database projection or Temporal schedule reconciliation. + +### 3. Context resolution is read-only + +The workflow first resolves and freezes context. Mutations then run in an +explicit bounded-operation stage and merge only normalized outcomes into the +snapshot before evidence and rule/instruction evaluation. + +SBOM selection remains a read in the context phase; its fixed selection is the +input to the operation stage. Package prune and backup have no discovery read +inside activity-core and bind a pending marker until their operation completes. + +### 4. Retry behavior is operation-specific + +- SBOM ingest uses stable per-run/per-repository idempotency keys and heartbeat + checkpoints, so Activity retries resume the frozen batch. +- Forgejo prune and CNPG backup have no activity-core-verifiable remote + idempotency receipt. Their operation activity therefore has one Temporal + attempt; a failure remains visible for operator reconciliation rather than + risking an automatic second mutation. + +This does not preclude future safe retries after the platform tools expose a +durable operation receipt. + +### 5. Evidence is mandatory and bounded + +Every operation must produce an allowlisted summary through a configured +report/evidence sink. Raw subprocess output, tokens, provider payloads, archive +URLs, kubeconfigs, and credential material are not evidence. + +## Rejected alternatives + +- **Keep mutating shell resolvers.** Rejected because resolution should be a + replayable read and the generic dispatcher hides mutation admission. +- **Generic command activity.** Rejected because command text/path from a + definition is remote code execution by configuration. +- **Move every operation to a rein.** Rejected for these fixed platform + operations; it adds an agent execution constellation without judgement or + repository work. Operations that exceed this ADR's bounds do belong there. +- **Remove all local operations.** Rejected because it recreates scattered cron + and loses Temporal/evidence guarantees for established maintenance. + +## Consequences + +- Definition parsing gains a central operation-policy validator. +- The workflow gains an explicit operation stage. +- The generic `shell` resolver becomes read-only. +- Existing definitions migrate without widening targets or permissions. +- The registry is intentionally small and architectural review is required to + expand it. diff --git a/docs/conventions.md b/docs/conventions.md index e119389..5ce7cf1 100644 --- a/docs/conventions.md +++ b/docs/conventions.md @@ -18,7 +18,6 @@ extension point `af654abb`). | Queue name | Registered workers | |---|---| | `orchestrator-tq` | `RunActivityWorkflow` and all its activities (`load_activity_definition`, `resolve_context`, `log_run`) | -| `task-execution-tq` | Legacy `TaskExecutorWorkflow` stub only when `ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB=true`; real execution belongs in per-repo workers / agent-harness | **Rule:** a workflow and its activities must be registered on the same task queue. Cross-queue activity calls require an explicit `task_queue` argument on @@ -32,7 +31,6 @@ See `docs/idempotency.md` for the full workflow ID strategy. Summary: - `RunActivityWorkflow`: `activity-{activity_id}:{trigger_key}` -- `TaskExecutorWorkflow`: `task-{run_id}:{task_type}:{index}` - Temporal Schedules: `activity-schedule-{activity_id}` --- @@ -56,15 +54,8 @@ Each worker process registers: - **Workflows**: `worker.register_workflow(WorkflowClass)` - **Activities**: `worker.register_activity(activity_function)` -A single process may run workers for multiple task queues, but each `Worker` -instance is bound to one task queue. Use separate `Worker` instances for -`orchestrator-tq` and `task-execution-tq`. - -`TaskExecutorWorkflow` is not a production execution surface for activity-core. -It exists only as a compatibility/idempotency stub that writes a synthetic -`task_instances` row in older tests and dev flows. Do not add concrete task -execution logic here; execution ownership belongs to per-repo workers or a -future execution-owned repo/workplan. +The activity-core process registers one `Worker` on `orchestrator-tq`. Execution +ownership belongs to downstream consumers and approved execution services. --- diff --git a/docs/execution-queue-boundary.md b/docs/execution-queue-boundary.md index 917002a..5c90013 100644 --- a/docs/execution-queue-boundary.md +++ b/docs/execution-queue-boundary.md @@ -39,6 +39,10 @@ ops_run being `open` is an **automation fire**. Those are different planes (ACT-ADR-005). Mapping State Hub `POST /execution/launch-requests` onto “create a workplan execution row here” would collapse them again. +The bounded-operation registry in ACT-ADR-007 is a separate, code-reviewed +maintenance exception. It neither consumes this queue nor turns an `ops_run` +into a workplan task. + ## Replacement for State Hub routes | State Hub route | Replacement | diff --git a/docs/idempotency.md b/docs/idempotency.md index 557a7d0..8074191 100644 --- a/docs/idempotency.md +++ b/docs/idempotency.md @@ -59,23 +59,10 @@ stable ID must be assigned one by the ingress boundary before entering the syste --- -## Task instance idempotency - -Each `TaskInstance` spawned by `RunActivityWorkflow` gets its own unique workflow ID: - -``` -task-{run_id}:{task_type}:{index} -``` - -This ensures that if `RunActivityWorkflow` is replayed by Temporal (e.g. after a worker -restart), it does not re-spawn task instances that were already started. - ---- - ## Database idempotency `activity_runs` uses `run_id` as the primary key (UUID). The `log_run` activity uses an upsert (`INSERT ... ON CONFLICT DO NOTHING`) so that Temporal activity retries -do not produce duplicate run records. - -`task_instances` similarly uses an upsert on `id`. +do not produce duplicate run records. Task emission idempotency uses the +consumer reference recorded in `task_spawn_log`; runtime delivery uses the +unique `ops_runs.idempotency_key`. diff --git a/docs/runbook.md b/docs/runbook.md index cbd02ee..9906101 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -427,11 +427,13 @@ Default: **`ISSUE_SINK_TYPE=state-hub`** (ACTIVITY-WP-0022). See | `null` | Dry-run | | `rest` | Intentional issue-core / external tracker only | -`TaskExecutorWorkflow` is **disabled** unless -`ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB=true` (legacy tests only). +The legacy `TaskExecutorWorkflow`, its feature flag, and `task_instances` table +were removed by ACTIVITY-WP-0035 after production inventory proved zero use. -`review_required` on instructions is **metadata only** until a downstream -review queue exists (issue-core / work-record lane) — see ACTIVITY-WP-0023-T09. +`review_advisory` on instructions is **advisory evidence only** and report +projections state `review_gate_applied=false`. Legacy `review_required` input is +normalized for compatibility; no pending-review queue or hold occurs here. See +ACT-ADR-003 and ACTIVITY-WP-0035. ## Ops run claim queue (ACTIVITY-WP-0026) @@ -736,7 +738,7 @@ the next scheduled fire is the next authoritative run. ### Multiple worker replicas Temporal workers are stateless and horizontally scalable. Run additional worker -processes to increase throughput on `orchestrator-tq` and `task-execution-tq`. +processes to increase throughput on `orchestrator-tq`. Each worker registers the same workflows/activities — Temporal distributes tasks across all pollers automatically. diff --git a/docs/task-emission-consumer-contract.md b/docs/task-emission-consumer-contract.md index 8d7c4ad..9ae0522 100644 --- a/docs/task-emission-consumer-contract.md +++ b/docs/task-emission-consumer-contract.md @@ -95,8 +95,8 @@ and still post the progress event — spawn without completion leaves `due=true` ## Anti-patterns -- Using `TaskExecutorWorkflow` in activity-core for real work (disabled by - default; ACTIVITY-WP-0023-T08). +- Adding an in-repo task executor or reviving the retired + `TaskExecutorWorkflow`/`task_instances` surface. - Global `ISSUE_SINK_TYPE=rest` for all definitions (reintroduces Forgejo spam). - Treating `task_spawn_log` or State Hub `activity_task_spawn` as claim authority (use `POST /ops-runs/claim` — ACT-ADR-005). diff --git a/e2e/tests/test_full_flow.py b/e2e/tests/test_full_flow.py index ccc0b8e..a8678ce 100644 --- a/e2e/tests/test_full_flow.py +++ b/e2e/tests/test_full_flow.py @@ -154,21 +154,6 @@ async def assert_run_in_db(run_id: str) -> bool: await engine.dispose() -async def assert_task_instances_in_db(run_id: str) -> bool: - engine = create_async_engine(ACTCORE_DB_URL) - try: - async with AsyncSession(engine) as session: - result = await session.execute( - sa.text("SELECT COUNT(*) FROM task_instances WHERE run_id = :rid"), - {"rid": run_id}, - ) - count = result.scalar() - ok = count > 0 - return step("TaskInstances written to DB", ok, f"count={count}") - finally: - await engine.dispose() - - # ── main ───────────────────────────────────────────────────────────────────── async def main() -> int: @@ -219,7 +204,6 @@ async def main() -> int: # 7. Assert DB results.append(await assert_run_in_db(run_id)) - results.append(await assert_task_instances_in_db(run_id)) finally: if worker_proc and worker_proc.poll() is None: diff --git a/history/2026-08-23-bounded-operation-compatibility-report.md b/history/2026-08-23-bounded-operation-compatibility-report.md new file mode 100644 index 0000000..c924973 --- /dev/null +++ b/history/2026-08-23-bounded-operation-compatibility-report.md @@ -0,0 +1,43 @@ +# Bounded-operation compatibility report — 2026-08-23 + +Scope: all 12 checked-in `activity-definitions/*.md` files plus the production +runtime projection in `k8s/railiance/20-runtime.yaml`. + +## Baseline classification + +The pre-enforcement inventory found three mutating definitions: + +| Definition | Previous source/query | Admitted operation | Bound | +| --- | --- | --- | --- | +| `daily-sbom-catchup` | `sbom-nexus/catch_up` | `sbom_nexus_ingest` | `apply` explicit; limit 1–3; mandatory `sbom_catchup` report sink | +| `weekly-forgejo-package-prune` | `shell/forgejo_package_prune` | `forgejo_package_prune` | canonical script; `apply` explicit; retain 1–10; protected live-image file; one Temporal attempt | +| `daily-cnpg-option-a-backup` | `shell/cnpg_option_a_backup` | `cnpg_option_a_backup` | canonical script; `dry_run` explicit; 1–10 named targets; timeout ≤7200s; one Temporal attempt | + +The other nine definitions contain no mutating source. Three named `shell` +queries remain registered as read-only discovery/report queries. No definition +contained definition-supplied command text or an unknown shell query. + +## Migration result + +The three definitions and their Kubernetes projection now declare an explicit +code-owned `operation` id. Strict parsing succeeds for all 12 definitions. +Negative tests prove refusal of unknown shell queries, missing/mismatched +operation ids, arbitrary script paths, absent evidence sinks, excessive target +or SBOM limits, and multiple operations in one definition. + +Package prune and CNPG backup dispatch moved out of the generic shell resolver. +Their activity result crosses the Temporal boundary only after operation-specific +allowlist projection; raw stdout/stderr and credential-shaped fields are not +workflow context or progress evidence. + +## Legacy executor production inventory + +Before removal, read-only production checks found: + +- `task_instances`: `0` rows; `max(created_at) = none`; +- worker deployment: no `ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB` variable; and +- Temporal visibility query for `WorkflowType="TaskExecutorWorkflow"`: `[]`. + +This met the ACTIVITY-WP-0035-T07 removal gate. Migration `0009` drops the empty +table and renames the spawn-audit column from `review_required` to +`review_advisory`. diff --git a/k8s/railiance/20-runtime.yaml b/k8s/railiance/20-runtime.yaml index 8252d2e..fc2d819 100644 --- a/k8s/railiance/20-runtime.yaml +++ b/k8s/railiance/20-runtime.yaml @@ -97,6 +97,7 @@ data: context_sources: - type: sbom-nexus query: catch_up + operation: sbom_nexus_ingest required: true params: limit: 3 @@ -120,7 +121,7 @@ data: prompt: | Deterministic SBOM catch-up report from context.catchup (no LLM). output_schema: "" - review_required: false + review_advisory: false report_sinks: - type: state-hub-progress event_type: sbom_catchup @@ -260,7 +261,7 @@ data: ] } output_schema: activity-core://schemas/daily-triage-report.json - review_required: false + review_advisory: false report_sinks: - type: working-memory path: custodian://memory/working diff --git a/migrations/versions/0009_drop_legacy_task_instances.py b/migrations/versions/0009_drop_legacy_task_instances.py new file mode 100644 index 0000000..e284a78 --- /dev/null +++ b/migrations/versions/0009_drop_legacy_task_instances.py @@ -0,0 +1,57 @@ +"""retire legacy executor and gate-sounding audit column + +Revision ID: 0009 +Revises: 0008 +Create Date: 2026-08-23 + +ACTIVITY-WP-0035-T07 — the disabled TaskExecutorWorkflow compatibility stub had +no production executions and task_instances had zero production rows. +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects.postgresql import JSONB + +revision: str = "0009" +down_revision: Union[str, Sequence[str], None] = "0008" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.drop_index("ix_task_instances_run_id", table_name="task_instances") + op.drop_table("task_instances") + op.alter_column( + "task_spawn_log", + "review_required", + new_column_name="review_advisory", + ) + + +def downgrade() -> None: + op.alter_column( + "task_spawn_log", + "review_advisory", + new_column_name="review_required", + ) + op.create_table( + "task_instances", + sa.Column("id", sa.UUID(), nullable=False), + sa.Column("run_id", sa.UUID(), nullable=False), + sa.Column("type", sa.Text(), nullable=False), + sa.Column("params", JSONB(), nullable=False, server_default="{}"), + sa.Column("status", sa.Text(), nullable=False, server_default="pending"), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("now()"), + ), + sa.ForeignKeyConstraint( + ["run_id"], ["activity_runs.run_id"], ondelete="CASCADE" + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_task_instances_run_id", "task_instances", ["run_id"]) diff --git a/schemas/activity-definition.json b/schemas/activity-definition.json index 6c6dfe9..20aa425 100644 --- a/schemas/activity-definition.json +++ b/schemas/activity-definition.json @@ -1,27 +1,124 @@ { "$defs": { - "ContextSource": { - "description": "Describes one external data source that the workflow queries to build\nthe context snapshot passed to evaluate_templates.", + "ActionDef": { "properties": { - "name": { - "description": "Logical name; referenced as 'context.' in task templates.", - "title": "Name", + "task_template": { + "description": "Path to task template .md, relative to repo root.", + "title": "Task Template", "type": "string" }, - "type": { - "description": "Source adapter type: 'db_query' | 'http_get' | 'static'.", - "title": "Type", + "target_repo": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "description": "Attribute-access expression or literal repo slug.", + "title": "Target Repo" + }, + "priority": { + "default": "medium", + "title": "Priority", "type": "string" }, - "config": { + "labels": { + "items": { + "type": "string" + }, + "title": "Labels", + "type": "array" + }, + "due_in_days": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Due In Days" + }, + "approach_hint": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Approach Hint" + }, + "harness_profile_ref": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Harness Profile Ref" + }, + "execution_refs": { "additionalProperties": true, - "description": "Source-specific configuration (SQL, URL, static value, etc.).", - "title": "Config", + "title": "Execution Refs", "type": "object" } }, "required": [ - "name", + "task_template" + ], + "title": "ActionDef", + "type": "object" + }, + "ContextSource": { + "description": "One external data source that the workflow queries to build the context snapshot.", + "properties": { + "name": { + "default": "", + "description": "Logical name; referenced as 'context.' in templates.", + "title": "Name", + "type": "string" + }, + "type": { + "description": "Source adapter type: 'repo-scoping' | 'state-hub' | etc.", + "title": "Type", + "type": "string" + }, + "query": { + "default": "", + "description": "Named query to execute against the source.", + "title": "Query", + "type": "string" + }, + "params": { + "additionalProperties": true, + "title": "Params", + "type": "object" + }, + "bind_to": { + "default": "", + "description": "Context key to bind the result to.", + "title": "Bind To", + "type": "string" + }, + "required": { + "default": false, + "description": "When true, resolver failures fail the activity run instead of binding {}.", + "title": "Required", + "type": "boolean" + } + }, + "required": [ "type" ], "title": "ContextSource", @@ -42,27 +139,40 @@ }, "timezone": { "default": "UTC", - "description": "IANA timezone name, e.g. 'Europe/Berlin'.", + "description": "IANA timezone name.", "title": "Timezone", "type": "string" }, "jitter_seconds": { "default": 0, - "description": "Maximum random delay (seconds) added to each trigger to spread load.", "minimum": 0, "title": "Jitter Seconds", "type": "integer" }, "misfire_policy": { "default": "skip", - "description": "skip: ignore any missed runs. catchup: replay missed runs up to a bounded limit. compress: run once covering the full missed window.", "enum": [ "skip", + "catchup_all", + "catchup_latest", "catchup", "compress" ], "title": "Misfire Policy", "type": "string" + }, + "catchup_window_seconds": { + "anyOf": [ + { + "minimum": 0, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Catchup Window Seconds" } }, "required": [ @@ -80,13 +190,13 @@ "type": "string" }, "event_type": { - "description": "Matches EventEnvelope.type. The router fires this activity when an event with this type is received.", + "description": "Matches EventEnvelope.type. Router fires this activity on match.", "title": "Event Type", "type": "string" }, "filters": { "additionalProperties": true, - "description": "Optional predicate filters applied to EventEnvelope.payload before routing. All filters must match for the activity to fire.", + "description": "All filters must match EventEnvelope.attributes for routing.", "title": "Filters", "type": "object" } @@ -97,11 +207,225 @@ "title": "EventTriggerConfig", "type": "object" }, + "InstructionDef": { + "properties": { + "id": { + "title": "Id", + "type": "string" + }, + "condition": { + "default": "", + "description": "Optional pre-filter using Rule DSL; empty means always execute.", + "title": "Condition", + "type": "string" + }, + "trusted_fields": { + "description": "Allowlist of event/context fields that may appear in the prompt template.", + "items": { + "type": "string" + }, + "title": "Trusted Fields", + "type": "array" + }, + "model": { + "description": "LLM model identifier, e.g. 'claude-sonnet-4-6'.", + "title": "Model", + "type": "string" + }, + "temperature": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Temperature" + }, + "max_tokens": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Max Tokens" + }, + "max_depth": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Max Depth" + }, + "model_params": { + "additionalProperties": true, + "title": "Model Params", + "type": "object" + }, + "prompt": { + "description": "Prompt template with {field.path} placeholders.", + "title": "Prompt", + "type": "string" + }, + "output_schema": { + "description": "Path to JSON Schema file for output validation.", + "title": "Output Schema", + "type": "string" + }, + "review_advisory": { + "default": false, + "description": "Advisory evidence only; activity-core applies no review gate.", + "title": "Review Advisory", + "type": "boolean" + }, + "review_required": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "default": null, + "deprecated": true, + "description": "Deprecated input alias for review_advisory.", + "title": "Review Required" + }, + "report_sinks": { + "items": { + "additionalProperties": true, + "type": "object" + }, + "title": "Report Sinks", + "type": "array" + }, + "approach_hint": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Approach Hint" + }, + "harness_profile_ref": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Harness Profile Ref" + }, + "execution_refs": { + "additionalProperties": true, + "title": "Execution Refs", + "type": "object" + } + }, + "required": [ + "id", + "trusted_fields", + "model", + "prompt", + "output_schema" + ], + "title": "InstructionDef", + "type": "object" + }, + "RuleDef": { + "properties": { + "id": { + "title": "Id", + "type": "string" + }, + "for_each": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "description": "Optional event/context path to a list for per-item rule expansion.", + "title": "For Each" + }, + "bind_as": { + "default": "item", + "description": "Context key used for each item when for_each is set.", + "title": "Bind As", + "type": "string" + }, + "condition": { + "default": "", + "description": "Rule DSL expression; empty string means always true.", + "title": "Condition", + "type": "string" + }, + "action": { + "$ref": "#/$defs/ActionDef" + } + }, + "required": [ + "id", + "action" + ], + "title": "RuleDef", + "type": "object" + }, + "ScheduledTriggerConfig": { + "description": "One-off future trigger that fires once at a specified UTC datetime.", + "properties": { + "trigger_type": { + "const": "scheduled", + "default": "scheduled", + "title": "Trigger Type", + "type": "string" + }, + "at": { + "description": "UTC datetime when the workflow should be triggered.", + "format": "date-time", + "title": "At", + "type": "string" + }, + "timezone": { + "default": "UTC", + "description": "IANA timezone name (informational).", + "title": "Timezone", + "type": "string" + } + }, + "required": [ + "at" + ], + "title": "ScheduledTriggerConfig", + "type": "object" + }, "TaskTemplate": { - "description": "Template for one task instance produced by RunActivityWorkflow.\n\nevaluate_templates() expands each template against the context snapshot\nto produce a concrete TaskInstance.", + "description": "Legacy persisted shape; no in-repo task executor consumes it.", "properties": { "task_type": { - "description": "Maps to a registered TaskExecutorWorkflow type, e.g. 'send_email'.", + "description": "Legacy downstream task type metadata; not an executor selector.", "title": "Task Type", "type": "string" }, @@ -115,12 +439,10 @@ } ], "default": null, - "description": "Optional Python expression evaluated against the context snapshot. Task is skipped if the expression is falsy. Example: \"context['user']['is_active'] == True\"", "title": "Condition" }, "params_template": { "additionalProperties": true, - "description": "Parameter template. String values starting with '{context.' are interpolated from the context snapshot at evaluation time.", "title": "Params Template", "type": "object" } @@ -132,31 +454,28 @@ "type": "object" } }, - "description": "Versioned definition of a single activity: its trigger, context resolution\nstrategy, and the task templates it can spawn.", + "description": "Versioned definition: trigger + context sources + rules/instructions.", "properties": { "id": { - "description": "Stable UUID. Used as the Temporal Schedule ID prefix (f'activity-schedule-{id}') and as the workflow ID component.", "format": "uuid", "title": "Id", "type": "string" }, "name": { - "description": "Human-readable name.", "title": "Name", "type": "string" }, "enabled": { "default": true, - "description": "When False the corresponding Temporal Schedule is paused and event routing is suppressed.", "title": "Enabled", "type": "boolean" }, "trigger_config": { - "description": "Cron or event trigger configuration.", "discriminator": { "mapping": { "cron": "#/$defs/CronTriggerConfig", - "event": "#/$defs/EventTriggerConfig" + "event": "#/$defs/EventTriggerConfig", + "scheduled": "#/$defs/ScheduledTriggerConfig" }, "propertyName": "trigger_type" }, @@ -166,6 +485,9 @@ }, { "$ref": "#/$defs/EventTriggerConfig" + }, + { + "$ref": "#/$defs/ScheduledTriggerConfig" } ], "title": "Trigger Config" @@ -177,6 +499,20 @@ "title": "Context Sources", "type": "array" }, + "rules": { + "items": { + "$ref": "#/$defs/RuleDef" + }, + "title": "Rules", + "type": "array" + }, + "instructions": { + "items": { + "$ref": "#/$defs/InstructionDef" + }, + "title": "Instructions", + "type": "array" + }, "task_templates": { "items": { "$ref": "#/$defs/TaskTemplate" @@ -186,7 +522,8 @@ }, "dedupe_key_strategy": { "default": "skip", - "description": "How to handle duplicate or missed trigger events. Should match CronTriggerConfig.misfire_policy for cron activities.", + "deprecated": true, + "description": "Legacy persistence compatibility metadata; runtime scheduling uses trigger_config.misfire_policy and this field does not deduplicate content", "enum": [ "skip", "catchup", @@ -197,10 +534,14 @@ }, "version": { "default": 1, - "description": "Incremented on breaking schema changes. Stored in activity_runs for audit purposes.", "minimum": 1, "title": "Version", "type": "integer" + }, + "status": { + "default": "active", + "title": "Status", + "type": "string" } }, "required": [ diff --git a/schemas/event-envelope.json b/schemas/event-envelope.json index 5690de4..65351d2 100644 --- a/schemas/event-envelope.json +++ b/schemas/event-envelope.json @@ -1,57 +1,45 @@ { - "description": "Standard internal event envelope. Every event, whether time-fired or\nbroker-delivered, is normalised into this shape before processing.", + "description": "Standard internal event envelope. All inbound events (NATS, webhook, cron)\nare normalised into this shape before processing.", "properties": { - "event_id": { - "description": "Stable unique ID. Used for deduplication: if an event with this ID has already been processed, the router skips it.", - "title": "Event Id", + "id": { + "description": "UUID v4 \u2014 stable unique ID for deduplication.", + "title": "Id", "type": "string" }, "type": { - "description": "Dot-namespaced event type, e.g. 'user.created'.", + "description": "Dot-namespaced event type, e.g. 'org.repo.registered'.", "title": "Type", "type": "string" }, - "source": { - "description": "Originating service or component, e.g. 'user-service'.", - "title": "Source", + "version": { + "default": "1.0", + "description": "Schema version string.", + "title": "Version", "type": "string" }, - "occurred_at": { + "timestamp": { "description": "When the event occurred (UTC).", "format": "date-time", - "title": "Occurred At", + "title": "Timestamp", "type": "string" }, - "subject": { - "description": "Primary resource affected, e.g. 'user/123'.", - "title": "Subject", + "publisher": { + "description": "Originating service, e.g. 'the-custodian/state-hub'.", + "title": "Publisher", "type": "string" }, - "trace_id": { - "description": "Distributed tracing correlation ID.", - "title": "Trace Id", - "type": "string" - }, - "schema_version": { - "default": "1.0", - "description": "Schema version string for forward-compatibility.", - "title": "Schema Version", - "type": "string" - }, - "payload": { + "attributes": { "additionalProperties": true, - "description": "Event-specific data; structure varies by event type.", - "title": "Payload", + "description": "Event-specific attributes; structure varies by event type.", + "title": "Attributes", "type": "object" } }, "required": [ - "event_id", + "id", "type", - "source", - "occurred_at", - "subject", - "trace_id" + "timestamp", + "publisher" ], "title": "EventEnvelope", "type": "object" diff --git a/src/activity_core/activities.py b/src/activity_core/activities.py index 632f846..7aef4b3 100644 --- a/src/activity_core/activities.py +++ b/src/activity_core/activities.py @@ -22,10 +22,11 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from temporalio import activity from temporalio.exceptions import ApplicationError +from activity_core.audit_projection import bounded_audit_projection from activity_core.db import make_engine from activity_core.issue_sink import get_issue_sink from activity_core.orm import ActivityDefinition as ActivityDefinitionRow -from activity_core.orm import ActivityRun, TaskInstance, TaskSpawnLog +from activity_core.orm import ActivityRun, TaskSpawnLog from activity_core.ops_run_queue import create_ops_run_from_spec from activity_core.llm_client import get_llm_client from activity_core.models import InstructionDef @@ -111,6 +112,16 @@ async def load_activity_definition(activity_id: str) -> dict: non_retryable=True, ) + from activity_core.bounded_operations import validate_bounded_operations + + try: + validate_bounded_operations(row.context_sources, row.instructions_json) + except ValueError as exc: + raise ApplicationError( + f"ActivityDefinition {activity_id!r} violates bounded-operation policy: {exc}", + non_retryable=True, + ) from exc + return { "id": str(row.id), "name": row.name, @@ -141,6 +152,11 @@ async def resolve_context( The 'static' type is handled inline without a registry entry. """ import activity_core.context_resolvers # noqa: F401 — registers all adapters + from activity_core.bounded_operations import ( + READ_ONLY_SHELL_QUERIES, + operation_spec_for_source, + pending_operation_value, + ) from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY snapshot: dict = {} @@ -156,6 +172,22 @@ async def resolve_context( # Strip the 'context.' namespace prefix so evaluator can find the key. bind_key = raw_bind.removeprefix("context.") if raw_bind.startswith("context.") else raw_bind + operation_spec = operation_spec_for_source(source) + if operation_spec is not None: + if source.get("operation") != operation_spec.operation_id: + raise ApplicationError( + f"Bounded operation {source_type!r}/{query!r} is not explicitly admitted", + non_retryable=True, + ) + if not operation_spec.resolve_before_execute: + snapshot[bind_key] = pending_operation_value(operation_spec) + continue + elif source_type == "shell" and query not in READ_ONLY_SHELL_QUERIES: + raise ApplicationError( + f"Shell query {query!r} is not registered as read-only", + non_retryable=True, + ) + if source_type == "static": value = source.get("config", {}).get("value") if isinstance(value, str) and ( @@ -199,6 +231,40 @@ async def resolve_context( return snapshot +@activity.defn +async def execute_bounded_operation(payload: dict[str, Any]) -> dict[str, Any]: + """Execute one code-registered non-SBOM operation and return its context patch.""" + from activity_core.bounded_operations import ( + normalize_operation_result, + operation_spec_for_source, + ) + + source = payload.get("source") + if not isinstance(source, dict): + raise ApplicationError("bounded operation source must be a mapping", non_retryable=True) + spec = operation_spec_for_source(source) + if spec is None or source.get("operation") != spec.operation_id: + raise ApplicationError("bounded operation is not registered", non_retryable=True) + params = source.get("params") or {} + if spec.operation_id == "forgejo_package_prune": + from activity_core.context_resolvers.forgejo_prune import forgejo_package_prune + + result = forgejo_package_prune(params) + elif spec.operation_id == "cnpg_option_a_backup": + from activity_core.context_resolvers.cnpg_backup import cnpg_option_a_backup + + result = cnpg_option_a_backup(params) + else: + raise ApplicationError( + f"bounded operation {spec.operation_id!r} uses a dedicated activity", + non_retryable=True, + ) + + raw_bind = source.get("bind_to") or source.get("name") or source.get("type") or "operation" + bind_key = str(raw_bind).removeprefix("context.") + return {bind_key: normalize_operation_result(spec.operation_id, result)} + + def _sbom_heartbeat_state(run_id: str) -> dict[str, Any]: try: details = activity.info().heartbeat_details @@ -307,7 +373,7 @@ async def log_run(run_payload: dict) -> str: activity_id=uuid.UUID(run_payload["activity_id"]), scheduled_for=scheduled_for, fired_at=datetime.now(tz=timezone.utc), - context_snapshot=run_payload["context_snapshot"], + context_snapshot=bounded_audit_projection(run_payload["context_snapshot"]), tasks_spawned=run_payload["tasks_spawned"], version_used=run_payload["version_used"], ) @@ -321,44 +387,6 @@ async def log_run(run_payload: dict) -> str: return str(run_id) -@activity.defn -async def persist_task_instance(task_payload: dict) -> str: - """Write a TaskInstance row and return its id. - - Idempotent: uses INSERT … ON CONFLICT (id) DO NOTHING. - - Expected keys in task_payload: - id (str UUID — deterministic, computed in TaskExecutorWorkflow) - run_id (str UUID) - type (str) - params (dict) - status (str, default "done" for stub) - - Returns: - task instance id as a str UUID. - """ - Session = _get_session_factory() - task_id = uuid.UUID(task_payload["id"]) - - stmt = ( - pg_insert(TaskInstance) - .values( - id=task_id, - run_id=uuid.UUID(task_payload["run_id"]), - type=task_payload["type"], - params=task_payload.get("params", {}), - status=task_payload.get("status", "done"), - ) - .on_conflict_do_nothing(index_elements=["id"]) - ) - - async with Session() as session: - async with session.begin(): - await session.execute(stmt) - - return str(task_id) - - @activity.defn async def evaluate_rules(payload: dict) -> list[dict]: """Evaluate rules and render matching actions as task specs. @@ -453,7 +481,7 @@ async def evaluate_instructions(payload: dict) -> dict: ) report = result.report output_validated = result.output_validated - review_required = result.review_required + review_advisory = result.review_advisory validation_error = result.validation_error # ACTIVITY-WP-0021-T05: when LLM produces nothing but a curated digest # is present and the instruction has report sinks, still emit a @@ -471,7 +499,7 @@ async def evaluate_instructions(payload: dict) -> dict: "digest_preview": digest[:4000], } output_validated = False - review_required = True + review_advisory = True validation_error = ( validation_error or "no_llm_report; posted deterministic digest" ) @@ -484,7 +512,8 @@ async def evaluate_instructions(payload: dict) -> dict: "prompt_hash": result.prompt_hash, "model": result.model, "output_validated": output_validated, - "review_required": review_required, + "review_advisory": review_advisory, + "review_gate_applied": False, "validation_error": validation_error, "llm_response_metadata": result.llm_response_metadata, }) @@ -502,7 +531,8 @@ async def evaluate_instructions(payload: dict) -> dict: "prompt_hash": result.prompt_hash, "model": result.model, "output_validated": result.output_validated, - "review_required": result.review_required, + "review_advisory": result.review_advisory, + "review_gate_applied": False, "approach_hint": instruction.approach_hint, "harness_profile_ref": instruction.harness_profile_ref, "execution_refs": instruction.execution_refs, @@ -636,14 +666,16 @@ async def emit_tasks(payload: dict) -> list[str]: activity_def_id=uuid.UUID(activity_id), source_type=spec.source_type, source_id=spec.source_id, - source_version="1", + source_version=str(payload.get("version_used", "1")), triggering_event_id=triggering_event_id, task_ref=ref.external_id, condition_matched=spec_dict.get("condition"), prompt_hash=spec_dict.get("prompt_hash"), model=spec_dict.get("model"), output_validated=spec_dict.get("output_validated"), - review_required=spec_dict.get("review_required"), + review_advisory=spec_dict.get( + "review_advisory", spec_dict.get("review_required") + ), ) session.add(log_row) except Exception as exc: diff --git a/src/activity_core/api.py b/src/activity_core/api.py index d9e7107..cb2ebac 100644 --- a/src/activity_core/api.py +++ b/src/activity_core/api.py @@ -1,6 +1,8 @@ """FastAPI REST API for activity-core. -T30: CRUD for ActivityDefinition + manual one-shot trigger. +T30: basic row administration for ActivityDefinition + manual one-shot trigger. +The REST schema does not round-trip markdown-authored rules or instructions; +use source sync for the complete definition contract. ACTIVITY-WP-0024: operator automation console under /ops. Endpoints: @@ -40,6 +42,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn from temporalio.api.workflowservice.v1 import GetSystemInfoRequest from temporalio.client import Client +from activity_core.bounded_operations import validate_bounded_operations from activity_core.models import ActivityDefinition, CronTriggerConfig from activity_core.execution_api import router as execution_router from activity_core.ops_api import bind_ops_deps, router as ops_router @@ -83,7 +86,14 @@ async def lifespan(app: FastAPI): # type: ignore[type-arg] await engine.dispose() -app = FastAPI(title="activity-core API", lifespan=lifespan) +app = FastAPI( + title="activity-core API", + description=( + "Automation operations plus basic ActivityDefinition row administration. " + "Markdown source sync owns the full rules/instructions contract." + ), + lifespan=lifespan, +) app.include_router(webhook_router) app.include_router(ops_router) app.include_router(ops_runs_router) @@ -152,6 +162,16 @@ def _row_to_response(row: ActivityDefinitionRow) -> ActivityDefinitionResponse: ) +def _validate_context_admission( + context_sources: list[dict[str, Any]], + instructions: list[dict[str, Any]] | None = None, +) -> None: + try: + validate_bounded_operations(context_sources, instructions or []) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + + async def _upsert_schedule_if_cron(row: ActivityDefinitionRow) -> None: """Upsert a Temporal Schedule for the row if it uses a cron trigger.""" try: @@ -197,7 +217,8 @@ async def get_definition(definition_id: uuid.UUID) -> ActivityDefinitionResponse @app.post("/activity-definitions/", response_model=ActivityDefinitionResponse, status_code=201) async def create_definition(body: ActivityDefinitionCreate) -> ActivityDefinitionResponse: - """Create a new ActivityDefinition. Upserts a Temporal Schedule if trigger_type='cron'.""" + """Create a basic definition row; source sync owns rules and instructions.""" + _validate_context_admission(body.context_sources) trigger_type = body.trigger_config.get("trigger_type", "") row = ActivityDefinitionRow( id=uuid.uuid4(), @@ -222,13 +243,20 @@ async def create_definition(body: ActivityDefinitionCreate) -> ActivityDefinitio async def update_definition( definition_id: uuid.UUID, body: ActivityDefinitionUpdate ) -> ActivityDefinitionResponse: - """Update an ActivityDefinition. Re-upserts the Temporal Schedule if trigger_type='cron'.""" + """Update basic row fields; source sync owns rules and instructions.""" Session = _get_db() async with Session() as session: row = await session.get(ActivityDefinitionRow, definition_id) if row is None: raise HTTPException(status_code=404, detail="ActivityDefinition not found") + _validate_context_admission( + body.context_sources + if body.context_sources is not None + else list(row.context_sources or []), + list(row.instructions_json or []), + ) + if body.name is not None: row.name = body.name if body.enabled is not None: diff --git a/src/activity_core/audit_projection.py b/src/activity_core/audit_projection.py new file mode 100644 index 0000000..20e6fdd --- /dev/null +++ b/src/activity_core/audit_projection.py @@ -0,0 +1,68 @@ +"""Bounded, non-secret projections for durable audit surfaces.""" + +from __future__ import annotations + +from typing import Any + +_SENSITIVE_KEYS = frozenset( + { + "api_key", + "api_token", + "access_token", + "refresh_token", + "authorization", + "client_secret", + "cookie", + "credential", + "credentials", + "messages", + "model_messages", + "password", + "private_key", + "provider_payload", + "provider_response", + "raw_output", + "raw_output_preview", + "raw_prompt", + "rendered_prompt", + "secret", + "token", + "tool_error", + "tool_output", + } +) + + +def bounded_audit_projection( + value: Any, + *, + max_depth: int = 8, + max_items: int = 100, + max_string: int = 4000, +) -> Any: + """Return a JSON-like audit projection without raw or credential fields. + + The projection is intentionally lossy. Workflow evaluation uses the full + in-memory context; only the durable audit copy is bounded here. + """ + + def project(item: Any, depth: int) -> Any: + if depth > max_depth: + return "" + if item is None or isinstance(item, (bool, int, float)): + return item + if isinstance(item, str): + return item[:max_string] + if isinstance(item, dict): + result: dict[str, Any] = {} + for raw_key, child in list(item.items())[:max_items]: + key = str(raw_key) + if key.strip().lower() in _SENSITIVE_KEYS: + continue + result[key] = project(child, depth + 1) + return result + if isinstance(item, (list, tuple)): + return [project(child, depth + 1) for child in item[:max_items]] + return str(item)[:max_string] + + return project(value, 0) diff --git a/src/activity_core/bounded_operations.py b/src/activity_core/bounded_operations.py new file mode 100644 index 0000000..ba9c9b1 --- /dev/null +++ b/src/activity_core/bounded_operations.py @@ -0,0 +1,292 @@ +"""Admission policy for ACT-ADR-007 bounded operations. + +Definitions may select only code-registered operations. This module validates +their safety envelope during markdown parsing and exposes immutable runtime +metadata to the workflow. It never accepts a command or import path from a +definition. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any, Iterable + + +@dataclass(frozen=True) +class BoundedOperationSpec: + operation_id: str + source_type: str + query: str + resolve_before_execute: bool + max_timeout_seconds: int + temporal_max_attempts: int + idempotency: str + credential_route: str + evidence_mode: str + + +_SPECS = ( + BoundedOperationSpec( + operation_id="sbom_nexus_ingest", + source_type="sbom-nexus", + query="catch_up", + resolve_before_execute=True, + max_timeout_seconds=900, + temporal_max_attempts=10, + idempotency="activity-run-id + repository; heartbeat acknowledged outcomes", + credential_route="activity-core-sbom-nexus", + evidence_mode="instruction-report:sbom_catchup", + ), + BoundedOperationSpec( + operation_id="forgejo_package_prune", + source_type="shell", + query="forgejo_package_prune", + resolve_before_execute=False, + max_timeout_seconds=900, + temporal_max_attempts=1, + idempotency="single Temporal attempt; operator reconciles ambiguous failure", + credential_route="forgejo-admin-api-token", + evidence_mode="context-evidence-sink", + ), + BoundedOperationSpec( + operation_id="cnpg_option_a_backup", + source_type="shell", + query="cnpg_option_a_backup", + resolve_before_execute=False, + max_timeout_seconds=7200, + temporal_max_attempts=1, + idempotency="single Temporal attempt; platform backup receipt is authoritative", + credential_route="railiance-cnpg-option-a-backup", + evidence_mode="context-evidence-sink", + ), +) + +BOUNDED_OPERATION_REGISTRY = {spec.operation_id: spec for spec in _SPECS} +_BY_SOURCE_QUERY = {(spec.source_type, spec.query): spec for spec in _SPECS} + +# The generic ``shell`` adapter is retained only as a compatibility namespace +# for these known read-only queries. Adding a name here requires code review; +# definition data cannot supply an arbitrary command. +READ_ONLY_SHELL_QUERIES = frozenset( + { + "reuse_surface_report_gaps", + "discover_kaizen_scheduled_repos", + "discover_kaizen_projects", + } +) + +_CANONICAL_SCRIPT_PATHS = { + "forgejo_package_prune": "/opt/railiance-platform/tools/cmd/forgejo-package-prune", + "cnpg_option_a_backup": "/opt/railiance-platform/tools/cmd/cnpg-option-a-backup", +} +_TARGET_RE = re.compile(r"^[A-Za-z0-9_.-]+$") + + +def operation_spec_for_source(source: dict[str, Any]) -> BoundedOperationSpec | None: + return _BY_SOURCE_QUERY.get((str(source.get("type") or ""), str(source.get("query") or ""))) + + +def operation_sources( + context_sources: Iterable[dict[str, Any]], +) -> list[tuple[dict[str, Any], BoundedOperationSpec]]: + found: list[tuple[dict[str, Any], BoundedOperationSpec]] = [] + for source in context_sources: + spec = operation_spec_for_source(source) + if spec is not None: + found.append((source, spec)) + return found + + +def pending_operation_value(spec: BoundedOperationSpec) -> dict[str, str]: + return {"operation": spec.operation_id, "status": "pending"} + + +def normalize_operation_result(operation_id: str, raw: Any) -> dict[str, Any]: + """Project subprocess output to the operation's non-secret evidence shape.""" + result = raw if isinstance(raw, dict) else {} + if operation_id == "forgejo_package_prune": + return { + "kind": operation_id, + "apply": bool(result.get("apply")), + "candidate_count": _safe_count(result.get("candidate_count")), + "deleted_count": _safe_count(result.get("deleted_count")), + "skipped_protected_count": _safe_count( + result.get("skipped_protected_count") + ), + "error_count": len(result.get("errors") or []) + if isinstance(result.get("errors"), list) + else 0, + } + if operation_id == "cnpg_option_a_backup": + return { + "kind": operation_id, + "overall": str(result.get("overall") or "unknown")[:40], + "dry_run": bool(result.get("dry_run")), + "dumped": _safe_count(result.get("dumped")), + "uploaded": _safe_count(result.get("uploaded")), + "failed": _safe_count(result.get("failed")), + "script_exit_code": _safe_count(result.get("script_exit_code")), + } + raise ValueError(f"no result projection for bounded operation {operation_id!r}") + + +def _safe_count(value: Any) -> int: + if isinstance(value, bool) or not isinstance(value, int): + return 0 + return max(0, value) + + +def validate_bounded_operations( + context_sources: list[dict[str, Any]], + instructions: list[dict[str, Any]], +) -> None: + """Fail closed on unknown shell queries and incomplete operation envelopes.""" + found = operation_sources(context_sources) + if len(found) > 1: + names = ", ".join(spec.operation_id for _, spec in found) + raise ValueError( + "an ActivityDefinition may declare at most one bounded operation; " + f"found: {names}" + ) + + for source in context_sources: + source_type = str(source.get("type") or "") + query = str(source.get("query") or "") + spec = operation_spec_for_source(source) + if source_type == "shell" and spec is None and query not in READ_ONLY_SHELL_QUERIES: + raise ValueError( + f"shell query {query!r} is not registered as read-only or as a " + "bounded operation" + ) + if spec is None: + if source.get("operation") is not None: + raise ValueError( + f"context source {source_type!r}/{query!r} declares unknown " + f"operation {source.get('operation')!r}" + ) + continue + + declared = source.get("operation") + if declared != spec.operation_id: + raise ValueError( + f"context source {source_type!r}/{query!r} must declare " + f"operation: {spec.operation_id}" + ) + params = source.get("params") + if not isinstance(params, dict): + raise ValueError(f"bounded operation {spec.operation_id} params must be a mapping") + _validate_timeout(spec, params) + + if spec.operation_id == "sbom_nexus_ingest": + _validate_sbom(params, instructions) + elif spec.operation_id == "forgejo_package_prune": + _validate_prune(params) + elif spec.operation_id == "cnpg_option_a_backup": + _validate_backup(params) + + +def _validate_timeout(spec: BoundedOperationSpec, params: dict[str, Any]) -> None: + raw = params.get("timeout_seconds", spec.max_timeout_seconds) + if isinstance(raw, bool) or not isinstance(raw, (int, float)): + raise ValueError(f"bounded operation {spec.operation_id} timeout_seconds must be numeric") + if raw <= 0 or raw > spec.max_timeout_seconds: + raise ValueError( + f"bounded operation {spec.operation_id} timeout_seconds must be in " + f"1..{spec.max_timeout_seconds}" + ) + + +def _explicit_bool(params: dict[str, Any], field: str, operation_id: str) -> bool: + if field not in params or not isinstance(params[field], bool): + raise ValueError( + f"bounded operation {operation_id} must declare boolean {field} explicitly" + ) + return bool(params[field]) + + +def _validate_sbom( + params: dict[str, Any], + instructions: list[dict[str, Any]], +) -> None: + _explicit_bool(params, "apply", "sbom_nexus_ingest") + limit = params.get("limit") + if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= 3: + raise ValueError("bounded operation sbom_nexus_ingest limit must be in 1..3") + if not _has_instruction_evidence(instructions, "sbom_catchup"): + raise ValueError( + "bounded operation sbom_nexus_ingest requires an instruction report " + "sink with event_type: sbom_catchup" + ) + + +def _validate_prune(params: dict[str, Any]) -> None: + apply = _explicit_bool(params, "apply", "forgejo_package_prune") + max_versions = params.get("max_versions") + if ( + isinstance(max_versions, bool) + or not isinstance(max_versions, int) + or not 1 <= max_versions <= 10 + ): + raise ValueError("bounded operation forgejo_package_prune max_versions must be in 1..10") + _validate_canonical_script(params, "prune_script", "forgejo_package_prune") + if apply and not str(params.get("live_images_file") or "").strip(): + raise ValueError( + "bounded operation forgejo_package_prune apply=true requires live_images_file" + ) + _require_context_evidence(params, "forgejo_package_prune") + + +def _validate_backup(params: dict[str, Any]) -> None: + _explicit_bool(params, "dry_run", "cnpg_option_a_backup") + _validate_canonical_script(params, "backup_script", "cnpg_option_a_backup") + raw_targets = params.get("targets") + if not isinstance(raw_targets, str): + raise ValueError("bounded operation cnpg_option_a_backup requires explicit targets") + targets = [item.strip() for item in raw_targets.split(",") if item.strip()] + if not targets or len(targets) > 10 or any(_TARGET_RE.fullmatch(item) is None for item in targets): + raise ValueError( + "bounded operation cnpg_option_a_backup targets must contain 1..10 safe names" + ) + _require_context_evidence(params, "cnpg_option_a_backup") + + +def _validate_canonical_script( + params: dict[str, Any], field: str, operation_id: str +) -> None: + expected = _CANONICAL_SCRIPT_PATHS[operation_id] + if params.get(field) != expected: + raise ValueError( + f"bounded operation {operation_id} {field} must be the canonical path {expected!r}" + ) + + +def _require_context_evidence(params: dict[str, Any], operation_id: str) -> None: + sinks = params.get("evidence_sinks") + if not isinstance(sinks, list) or not any( + isinstance(item, dict) + and item.get("type") == "state-hub-progress" + and item.get("event_type") == operation_id + for item in sinks + ): + raise ValueError( + f"bounded operation {operation_id} requires a state-hub-progress " + f"evidence sink with event_type: {operation_id}" + ) + + +def _has_instruction_evidence( + instructions: list[dict[str, Any]], event_type: str +) -> bool: + for instruction in instructions: + sinks = instruction.get("report_sinks") + if not isinstance(sinks, list): + continue + if any( + isinstance(sink, dict) + and sink.get("type") == "state-hub-progress" + and sink.get("event_type") == event_type + for sink in sinks + ): + return True + return False diff --git a/src/activity_core/context_resolvers/reuse_surface.py b/src/activity_core/context_resolvers/reuse_surface.py index e590d27..e4df7f1 100644 --- a/src/activity_core/context_resolvers/reuse_surface.py +++ b/src/activity_core/context_resolvers/reuse_surface.py @@ -21,8 +21,6 @@ import httpx import yaml from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY, ContextResolver -from activity_core.context_resolvers.forgejo_prune import forgejo_package_prune -from activity_core.context_resolvers.cnpg_backup import cnpg_option_a_backup from activity_core.context_resolvers.kaizen import KaizenContextResolver from activity_core.context_resolvers.state_hub import StateHubContextResolver @@ -506,15 +504,15 @@ class ReuseSurfaceContextResolver(ContextResolver): class ShellContextResolver(ContextResolver): - """Dispatch shell-backed context queries without breaking kaizen aliases.""" + """Dispatch code-registered read-only shell compatibility queries.""" def resolve(self, query: str, event: Any, params: dict[str, Any]) -> dict[str, Any]: if query == "reuse_surface_report_gaps": return reuse_surface_report_gaps(params) - if query == "forgejo_package_prune": - return forgejo_package_prune(params) - if query == "cnpg_option_a_backup": - return cnpg_option_a_backup(params) + if query in {"forgejo_package_prune", "cnpg_option_a_backup"}: + raise RuntimeError( + f"mutating query {query!r} must run through the bounded-operation stage" + ) return KaizenContextResolver().resolve(query, event, params) diff --git a/src/activity_core/definition_parser.py b/src/activity_core/definition_parser.py index 7941f71..a49b9cb 100644 --- a/src/activity_core/definition_parser.py +++ b/src/activity_core/definition_parser.py @@ -17,6 +17,7 @@ from typing import Any import yaml +from activity_core.bounded_operations import validate_bounded_operations from activity_core.glas_profile import ( ProfileRefError, require_harness_profile, @@ -122,6 +123,28 @@ def _validate_execution_declarations( ) +def _normalise_review_advisory( + instructions: list[dict[str, Any]], file: Path +) -> None: + """Migrate the legacy gate-sounding field to advisory-only semantics.""" + for instruction in instructions: + legacy_present = "review_required" in instruction + advisory_present = "review_advisory" in instruction + if legacy_present and advisory_present and ( + bool(instruction["review_required"]) + != bool(instruction["review_advisory"]) + ): + raise ParseError( + file, + None, + f"instruction {instruction.get('id')!r} declares conflicting " + "review_required and review_advisory values", + ) + if legacy_present: + instruction.setdefault("review_advisory", bool(instruction["review_required"])) + instruction.pop("review_required", None) + + def _scan_dirs() -> list[Path]: dirs: list[Path] = [] default_dir = Path("activity-definitions") @@ -236,7 +259,12 @@ def parse_file(path: Path) -> ActivityDefinitionDef: raise ParseError(path, None, "instruction block missing required field 'id'") instructions.append(block_data) + _normalise_review_advisory(instructions, path) _validate_execution_declarations(rules, instructions, path) + try: + validate_bounded_operations(context_sources, instructions) + except ValueError as exc: + raise ParseError(path, None, str(exc)) from exc return ActivityDefinitionDef( id=str(fm["id"]), diff --git a/src/activity_core/models.py b/src/activity_core/models.py index 94038a3..6d8a617 100644 --- a/src/activity_core/models.py +++ b/src/activity_core/models.py @@ -9,7 +9,7 @@ from typing import Annotated, Any, Literal, Union from datetime import datetime from uuid import UUID -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator # ── EventEnvelope (T40) ─────────────────────────────────────────────────────── @@ -137,12 +137,34 @@ class InstructionDef(BaseModel): model_params: dict[str, Any] = Field(default_factory=dict) prompt: str = Field(description="Prompt template with {field.path} placeholders.") output_schema: str = Field(description="Path to JSON Schema file for output validation.") - review_required: bool = Field(default=False) + review_advisory: bool = Field( + default=False, + description="Advisory evidence only; activity-core applies no review gate.", + ) + review_required: bool | None = Field( + default=None, + exclude=True, + repr=False, + description="Deprecated input alias for review_advisory.", + json_schema_extra={"deprecated": True}, + ) 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) + @model_validator(mode="before") + @classmethod + def _accept_legacy_review_required(cls, data: Any) -> Any: + if not isinstance(data, dict) or "review_required" not in data: + return data + normalized = dict(data) + legacy = bool(normalized["review_required"]) + if "review_advisory" in normalized and bool(normalized["review_advisory"]) != legacy: + raise ValueError("review_required conflicts with review_advisory") + normalized["review_advisory"] = legacy + return normalized + # ── Context sources ─────────────────────────────────────────────────────────── @@ -166,9 +188,11 @@ class ContextSource(BaseModel): # ── Task templates (legacy) ─────────────────────────────────────────────────── class TaskTemplate(BaseModel): - """Legacy task template — ignored when ActivityDefinition.rules is non-empty.""" + """Legacy persisted shape; no in-repo task executor consumes it.""" - task_type: str + task_type: str = Field( + description="Legacy downstream task type metadata; not an executor selector." + ) condition: str | None = None params_template: dict[str, Any] = Field(default_factory=dict) @@ -186,7 +210,7 @@ class ActivityDefinition(BaseModel): # New rule/instruction pipeline (T34) rules: list[RuleDef] = Field(default_factory=list) instructions: list[InstructionDef] = Field(default_factory=list) - # Legacy — ignored when rules is non-empty + # Legacy persisted/API compatibility; RunActivityWorkflow does not execute it. task_templates: list[TaskTemplate] = Field(default_factory=list) dedupe_key_strategy: Literal["skip", "catchup", "compress"] = Field( default="skip", diff --git a/src/activity_core/ops_evidence_sinks.py b/src/activity_core/ops_evidence_sinks.py index 567521a..2bd2b52 100644 --- a/src/activity_core/ops_evidence_sinks.py +++ b/src/activity_core/ops_evidence_sinks.py @@ -11,6 +11,7 @@ from uuid import NAMESPACE_URL, UUID, uuid5 import httpx +from activity_core.audit_projection import bounded_audit_projection from activity_core.context_resolvers.ops_inventory import _sanitize_url from activity_core.state_hub_write import ( apply_progress_scope_fields, @@ -55,7 +56,9 @@ def persist_ops_inventory_evidence(payload: dict[str, Any]) -> list[dict[str, An continue bind_key = _context_bind_key(source) - probe_result = (payload.get("context") or {}).get(bind_key) + probe_result = bounded_audit_projection( + (payload.get("context") or {}).get(bind_key) + ) if isinstance(probe_result, dict) and probe_result.get("skipped"): results.append({ "type": "state-hub-progress", @@ -685,8 +688,8 @@ def _forgejo_package_prune_summary_text(result: dict[str, Any]) -> str: protected = result.get("skipped_protected_count", 0) apply = result.get("apply", False) mode = "apply" if apply else "dry-run" - errors = result.get("errors") or [] - error_note = f"; {len(errors)} error(s)" if errors else "" + error_count = result.get("error_count", 0) + error_note = f"; {error_count} error(s)" if error_count else "" return ( f"Forgejo package prune ({mode}): {deleted} deleted, " f"{candidates} candidate(s), {protected} protected skip(s){error_note}" diff --git a/src/activity_core/orm.py b/src/activity_core/orm.py index 66ffd57..878f235 100644 --- a/src/activity_core/orm.py +++ b/src/activity_core/orm.py @@ -99,7 +99,7 @@ class TaskSpawnLog(Base): prompt_hash: Mapped[str | None] = mapped_column(String(64), nullable=True) model: Mapped[str | None] = mapped_column(Text, nullable=True) output_validated: Mapped[bool | None] = mapped_column(Boolean, nullable=True) - review_required: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + review_advisory: Mapped[bool | None] = mapped_column(Boolean, nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) @@ -120,26 +120,6 @@ class EventType(Base): ) -class TaskInstance(Base): - __tablename__ = "task_instances" - - id: Mapped[uuid.UUID] = mapped_column( - UUID(as_uuid=True), primary_key=True, default=uuid.uuid4 - ) - run_id: Mapped[uuid.UUID] = mapped_column( - UUID(as_uuid=True), - ForeignKey("activity_runs.run_id", ondelete="CASCADE"), - nullable=False, - index=True, - ) - type: Mapped[str] = mapped_column(Text, nullable=False) - params: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) - status: Mapped[str] = mapped_column(Text, nullable=False, default="pending") - created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), nullable=False, server_default=func.now() - ) - - class OpsRun(Base): """Claimable automation run instance (ACT-ADR-005 / ACTIVITY-WP-0026).""" diff --git a/src/activity_core/report_sinks.py b/src/activity_core/report_sinks.py index 06e8547..bccee21 100644 --- a/src/activity_core/report_sinks.py +++ b/src/activity_core/report_sinks.py @@ -11,6 +11,7 @@ from zoneinfo import ZoneInfo import httpx +from activity_core.audit_projection import bounded_audit_projection from activity_core.runtime_paths import ( custodian_repo_relative, custodian_repo_root, @@ -34,7 +35,7 @@ def persist_reports(payload: dict[str, Any]) -> list[dict[str, Any]]: """ results: list[dict[str, Any]] = [] for report_entry in payload.get("reports", []): - report_context = dict(report_entry) + report_context = bounded_audit_projection(dict(report_entry)) for sink in report_entry.get("sinks", []): sink_type = sink.get("type") try: @@ -139,7 +140,10 @@ def _post_state_hub_progress( "instruction_id": instruction_id, "scheduled_for": payload.get("scheduled_for"), "output_validated": report_entry.get("output_validated"), - "review_required": report_entry.get("review_required"), + "review_advisory": report_entry.get( + "review_advisory", report_entry.get("review_required") + ), + "review_gate_applied": False, "validation_error": report_entry.get("validation_error"), "llm_response_metadata": report_entry.get("llm_response_metadata"), "report": report, @@ -222,7 +226,8 @@ def _render_markdown( f"instruction_id: {instruction_id}", f"scheduled_for: {payload.get('scheduled_for')}", f"output_validated: {str(bool(report_entry.get('output_validated'))).lower()}", - f"review_required: {str(bool(report_entry.get('review_required'))).lower()}", + f"review_advisory: {str(bool(report_entry.get('review_advisory', report_entry.get('review_required')))).lower()}", + "review_gate_applied: false", f"model: {report_entry.get('model') or ''}", f"prompt_hash: {report_entry.get('prompt_hash') or ''}", f"created: {datetime.now(tz=timezone.utc).isoformat()}", diff --git a/src/activity_core/rules/executor.py b/src/activity_core/rules/executor.py index 556ce90..1679e14 100644 --- a/src/activity_core/rules/executor.py +++ b/src/activity_core/rules/executor.py @@ -38,12 +38,20 @@ class InstructionResult: prompt_hash: str | None = None model: str | None = None output_validated: bool = False - review_required: bool = False + review_advisory: bool = False condition_matched: str | None = None validation_error: str | None = None llm_response_metadata: dict[str, Any] | None = None +def _review_advisory(instr: Any) -> bool: + """Read the renamed advisory flag while accepting legacy caller objects.""" + legacy = getattr(instr, "review_required", None) + if legacy is not None: + return bool(legacy) + return bool(getattr(instr, "review_advisory", False)) + + def _resolve_path(obj: Any, path: str) -> Any: """Walk a dot-separated path on obj or dict. Returns None if not found.""" parts = path.split(".") @@ -134,7 +142,7 @@ def execute_instruction_with_audit( prompt_hash=None, model=getattr(instr, "model", None), output_validated=False, - review_required=True, + review_advisory=True, condition_matched=getattr(instr, "condition", "") or None, validation_error=str(exc), ) @@ -149,7 +157,7 @@ def execute_instruction_with_audit( prompt_hash=None, model=getattr(instr, "model", None), output_validated=False, - review_required=True, + review_advisory=True, condition_matched=getattr(instr, "condition", "") or None, validation_error=str(exc), ) @@ -200,13 +208,10 @@ def _execute( response_metadata = _llm_response_metadata(llm_client) task_specs, report, error = _validate_output(raw_output, instr, allow_list) if error: - # Truncate to keep log volume bounded but long enough to see the - # actual JSON shape mismatch (typical reports are <2KB). - preview = (raw_output or "")[:2000] logger.warning( "instruction_output_error: instruction=%r, prompt_hash=%s, " - "error=%s, raw_output_preview=%r", - instr.id, prompt_hash, error, preview, + "error=%s", + instr.id, prompt_hash, error, ) # Posture B (WP-0016-T03): try to recover a partial-but-usable # report from individually-parseable items before declaring total @@ -227,7 +232,7 @@ def _execute( prompt_hash=prompt_hash, model=instr.model, output_validated=False, - review_required=True, + review_advisory=True, condition_matched=instr.condition or None, validation_error=error, llm_response_metadata=response_metadata, @@ -240,7 +245,7 @@ def _execute( prompt_hash=prompt_hash, model=instr.model, output_validated=True, - review_required=bool(getattr(instr, "review_required", False)), + review_advisory=_review_advisory(instr), condition_matched=instr.condition or None, llm_response_metadata=response_metadata, ) @@ -272,7 +277,7 @@ def _empty_result( prompt_hash=prompt_hash, model=getattr(instr, "model", None), output_validated=False, - review_required=bool(getattr(instr, "review_required", False)), + review_advisory=_review_advisory(instr), condition_matched=getattr(instr, "condition", "") or None, validation_error=validation_error, ) @@ -281,47 +286,28 @@ def _empty_result( def _invalid_output_report( instr: Any, validation_error: str, - raw_output: Any, + _raw_output: Any, response_metadata: dict[str, Any] | None = None, ) -> dict[str, Any] | None: """Build a durable diagnostic report for invalid report-sink output. Task-only instructions keep the legacy empty-result behavior. Instructions - with report sinks should leave operators a bounded artifact that preserves - the partial model output without marking it as schema-valid. + with report sinks leave a bounded diagnostic without retaining unvalidated + provider output. """ if not getattr(instr, "report_sinks", None): return None - partial_output: Any - raw_preview: str | None = None - if isinstance(raw_output, str): - try: - partial_output = _parse_json_output(raw_output) - except json.JSONDecodeError: - partial_output = None - raw_preview = raw_output[:_RAW_OUTPUT_PREVIEW_LIMIT] - else: - partial_output = raw_output - report: dict[str, Any] = { "summary": ( f"Instruction {instr.id} produced output that failed validation; " - "partial output was preserved for operator review." + "unvalidated provider output was discarded." ), "status": "validation_failed", "validation_error": validation_error, } if response_metadata: report["llm_response_metadata"] = response_metadata - if isinstance(partial_output, dict): - if isinstance(partial_output.get("summary"), str): - report["partial_summary"] = partial_output["summary"] - report["partial_report"] = partial_output - elif isinstance(partial_output, list): - report["partial_report"] = partial_output - elif raw_preview is not None: - report["raw_output_preview"] = raw_preview return report @@ -343,7 +329,6 @@ _SNIPPET_LIMIT = 200 # fail the whole report or flow unbounded into a downstream consumer. _MAX_STRING_LEN = 4000 _MAX_DEPTH = 8 -_RAW_OUTPUT_PREVIEW_LIMIT = 12000 _SUMMARY_RE = re.compile(r'"summary"\s*:\s*"((?:[^"\\]|\\.)*)"') @@ -670,7 +655,7 @@ def _resilient_report( prompt_hash=prompt_hash, model=getattr(instr, "model", None), output_validated=True, - review_required=True, + review_advisory=True, condition_matched=getattr(instr, "condition", "") or None, validation_error=None, llm_response_metadata=response_metadata, @@ -745,7 +730,7 @@ def _deterministic_context_report(instr: Any, context: dict) -> InstructionResul prompt_hash=None, model=getattr(instr, "model", None), output_validated=True, - review_required=bool(getattr(instr, "review_required", False)), + review_advisory=_review_advisory(instr), condition_matched=getattr(instr, "condition", "") or None, ) @@ -806,7 +791,7 @@ def _sbom_catchup_report(instr: Any, catchup: dict) -> InstructionResult: prompt_hash=None, model=getattr(instr, "model", None), output_validated=True, - review_required=bool(getattr(instr, "review_required", False)), + review_advisory=_review_advisory(instr), condition_matched=getattr(instr, "condition", "") or None, ) diff --git a/src/activity_core/run_artifacts.py b/src/activity_core/run_artifacts.py index 318e73c..cbf55d1 100644 --- a/src/activity_core/run_artifacts.py +++ b/src/activity_core/run_artifacts.py @@ -15,6 +15,7 @@ from urllib.parse import quote from sqlalchemy import Select, select from sqlalchemy.ext.asyncio import AsyncSession +from activity_core.audit_projection import bounded_audit_projection from activity_core.glas_evidence import normalise_ops_result from activity_core.orm import ActivityRun, OpsRun, TaskSpawnLog @@ -37,6 +38,14 @@ def forgejo_org() -> str: ) +def public_context_keys(context_snapshot: Any) -> list[str]: + """Return safe top-level context names for the public run projection.""" + projected = bounded_audit_projection(context_snapshot or {}) + if not isinstance(projected, dict): + return [] + return sorted(projected.keys())[:40] + + def build_forgejo_blob_url( *, target_repo: str | None, @@ -310,7 +319,7 @@ async def enrich_activity_runs( "artifacts": artifacts[:12], "evidence": { "task_spawns": spawns[:20], - "context_keys": sorted((r.context_snapshot or {}).keys())[:40], + "context_keys": public_context_keys(r.context_snapshot), }, } ) diff --git a/src/activity_core/worker.py b/src/activity_core/worker.py index e98a5cf..39f30d7 100644 --- a/src/activity_core/worker.py +++ b/src/activity_core/worker.py @@ -1,8 +1,6 @@ """Temporal worker entrypoint for activity-core. -Starts two workers (wired up in T20): - - orchestrator-tq: RunActivityWorkflow + its activities - - task-execution-tq: TaskExecutorWorkflow +Starts the ``orchestrator-tq`` worker for RunActivityWorkflow and its activities. T23: Calls sync_schedules before entering the worker run loop to ensure all cron ActivityDefinitions have live Temporal Schedules. @@ -35,6 +33,7 @@ from temporalio.worker import Worker from activity_core.activities import ( apply_sbom_catchup, emit_tasks, + execute_bounded_operation, evaluate_instructions, evaluate_rules, init_session_factory, @@ -42,13 +41,12 @@ from activity_core.activities import ( log_run, persist_instruction_reports, persist_ops_evidence, - persist_task_instance, resolve_context, ) from activity_core.db import make_engine from sqlalchemy.ext.asyncio import async_sessionmaker from activity_core.sync_service import run_sync -from activity_core.workflows import RunActivityWorkflow, TaskExecutorWorkflow +from activity_core.workflows import RunActivityWorkflow logger = logging.getLogger(__name__) @@ -57,7 +55,6 @@ TEMPORAL_NAMESPACE = os.environ.get("TEMPORAL_NAMESPACE", "default") PROMETHEUS_BIND_ADDR = os.environ.get("PROMETHEUS_BIND_ADDR", "0.0.0.0:9090") ORCHESTRATOR_TASK_QUEUE = "orchestrator-tq" -TASK_EXECUTION_TASK_QUEUE = "task-execution-tq" async def run() -> None: @@ -106,6 +103,7 @@ async def run() -> None: load_activity_definition, resolve_context, apply_sbom_catchup, + execute_bounded_operation, log_run, evaluate_rules, evaluate_instructions, @@ -115,51 +113,15 @@ async def run() -> None: ], ) - # ACTIVITY-WP-0023-T08: only register the legacy task-execution stub when - # explicitly enabled. Default is orchestrator-only so production does not - # advertise a fake execution surface on task-execution-tq. - enable_task_stub = os.environ.get( - "ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB", "" - ).strip().lower() in {"1", "true", "yes", "on"} - - task_worker: Worker | None = None - if enable_task_stub: - task_worker = Worker( - client, - task_queue=TASK_EXECUTION_TASK_QUEUE, - workflows=[TaskExecutorWorkflow], - activities=[persist_task_instance], - ) - logger.warning( - "TaskExecutorWorkflow stub ENABLED on %s — not for production execution", - TASK_EXECUTION_TASK_QUEUE, - ) - else: - logger.info( - "TaskExecutorWorkflow stub not registered " - "(set ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB=true for legacy tests only)" - ) - loop = asyncio.get_running_loop() stop = asyncio.Event() loop.add_signal_handler(signal.SIGTERM, stop.set) loop.add_signal_handler(signal.SIGINT, stop.set) - workers = [orchestrator_worker] - if task_worker is not None: - workers.append(task_worker) - - from contextlib import AsyncExitStack - - async with AsyncExitStack() as stack: - for w in workers: - await stack.enter_async_context(w) - queues = [ORCHESTRATOR_TASK_QUEUE] - if enable_task_stub: - queues.append(TASK_EXECUTION_TASK_QUEUE) + async with orchestrator_worker: logger.info( "Workers running — queues: %r (namespace=%r)", - queues, + [ORCHESTRATOR_TASK_QUEUE], TEMPORAL_NAMESPACE, ) await stop.wait() diff --git a/src/activity_core/workflows.py b/src/activity_core/workflows.py index f112871..de04de1 100644 --- a/src/activity_core/workflows.py +++ b/src/activity_core/workflows.py @@ -1,12 +1,7 @@ """Temporal workflow definitions for activity-core. -Two workflows are registered here: - - RunActivityWorkflow → orchestrator-tq - - TaskExecutorWorkflow → task-execution-tq - -Workflow IDs follow the conventions in docs/conventions.md: - RunActivityWorkflow: activity-{activity_id}:{trigger_key} - TaskExecutorWorkflow: task-{run_id}:{task_type}:{index} +RunActivityWorkflow is registered on ``orchestrator-tq``. Workflow IDs follow +``activity-{activity_id}:{trigger_key}``; see docs/conventions.md. """ from __future__ import annotations @@ -22,15 +17,16 @@ with workflow.unsafe.imports_passed_through(): from activity_core.activities import ( apply_sbom_catchup, emit_tasks, + execute_bounded_operation, evaluate_rules, evaluate_instructions, load_activity_definition, log_run, persist_instruction_reports, persist_ops_evidence, - persist_task_instance, resolve_context, ) + from activity_core.bounded_operations import operation_sources from activity_core.ops_run_queue import emit_triggering_event_id from activity_core.schedule_manager import SCHEDULED_TRIGGER_KEY @@ -49,7 +45,7 @@ _RETRY_POLICY = RetryPolicy( _ACTIVITY_TIMEOUT = timedelta( seconds=int(os.environ.get("ACTIVITY_TIMEOUT_SECONDS", "900")) ) -_TASK_QUEUE = "task-execution-tq" +_NO_RETRY_POLICY = RetryPolicy(maximum_attempts=1) @workflow.defn @@ -59,9 +55,10 @@ class RunActivityWorkflow: Sequence: 1. load_activity_definition(activity_id) → defn dict 2. resolve_context(defn.context_sources) → read-only context snapshot - 3. apply_sbom_catchup(fixed selection) → bounded outcome patch - 4. evaluate rules/instructions → TaskSpec dicts and reports - 5. log run, then emit tasks + 3. execute admitted bounded operations → bounded outcome patches + 4. persist operation evidence + 5. evaluate rules/instructions → TaskSpec dicts and reports + 6. log run, then emit tasks """ @workflow.run @@ -114,7 +111,11 @@ class RunActivityWorkflow: retry_policy=_RETRY_POLICY, ) - # ── 3. Apply declared bounded side-effects to the fixed selection ──── + # ── 3. Execute admitted bounded operations ────────────────────────── + # SBOM retains its dedicated heartbeat/retry activity because it has a + # stable remote idempotency key. Shell-backed operations are one-shot: + # their platform tools do not expose a receipt activity-core can use to + # prove an ambiguous mutation safe to repeat. if any( isinstance(source, dict) and source.get("type") == "sbom-nexus" @@ -138,6 +139,22 @@ class RunActivityWorkflow: if isinstance(current, dict) and isinstance(patch, dict): current.update(patch) + for source, operation_spec in operation_sources( + defn.get("context_sources", []) + ): + if operation_spec.operation_id == "sbom_nexus_ingest": + continue + operation_patch: dict = await workflow.execute_activity( + execute_bounded_operation, + {"source": source, "run_id": run_id}, + start_to_close_timeout=timedelta( + seconds=operation_spec.max_timeout_seconds + ), + retry_policy=_NO_RETRY_POLICY, + ) + context_snapshot.update(operation_patch) + + # ── 4. Persist bounded operation/read evidence ─────────────────────── await workflow.execute_activity( persist_ops_evidence, { @@ -152,7 +169,7 @@ class RunActivityWorkflow: retry_policy=_RETRY_POLICY, ) - # ── 4. Evaluate rules ───────────────────────────────────────────────── + # ── 5. Evaluate rules ───────────────────────────────────────────────── import json as _json event_attrs: dict = {} if event_envelope_json: @@ -187,7 +204,7 @@ class RunActivityWorkflow: task_spec_dicts.extend(instruction_result.get("task_specs", [])) report_dicts.extend(instruction_result.get("reports", [])) - # ── 5. Persist reports ──────────────────────────────────────────────── + # ── 6. Persist reports ──────────────────────────────────────────────── if report_dicts: await workflow.execute_activity( persist_instruction_reports, @@ -202,7 +219,7 @@ class RunActivityWorkflow: retry_policy=_RETRY_POLICY, ) - # ── 6. Log the run BEFORE emit ──────────────────────────────────────── + # ── 7. Log the run BEFORE emit ──────────────────────────────────────── # ACTIVITY-WP-0021: emit_tasks sink failures used to abort the workflow # before log_run, so failed Binky/SBOM fires left no activity_runs row # and automation-status could not observe them. Always record the run; @@ -221,7 +238,7 @@ class RunActivityWorkflow: retry_policy=_RETRY_POLICY, ) - # ── 7. Emit tasks (may fail independently of run audit) ─────────────── + # ── 8. Emit tasks (may fail independently of run audit) ─────────────── if task_spec_dicts: # Cron schedules pass trigger_key="scheduled" for *every* fire. # ops_run idempotency is {def}:{source}:{triggering_event_id}, so @@ -239,70 +256,10 @@ class RunActivityWorkflow: "activity_id": activity_id, "triggering_event_id": emit_trigger_id, "run_id": run_id, + "version_used": defn["version"], }, start_to_close_timeout=_ACTIVITY_TIMEOUT, retry_policy=_RETRY_POLICY, ) return {"run_id": run_id, "tasks_spawned": len(task_spec_dicts)} - - -@workflow.defn -class TaskExecutorWorkflow: - """LEGACY NO-OP — not a production execution surface (ACTIVITY-WP-0023-T08). - - Historical compatibility stub. Real task execution belongs in per-repo - workers / agent-harness, not activity-core. - - Behaviour is controlled by ``ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB``: - - - unset / false (default): refuse to run (logs error, raises) so the stub - cannot attract production work by accident. - - true: legacy behaviour — persist a done ``task_instances`` row for - idempotent dev/test callers only. - """ - - @workflow.run - async def run(self, run_id: str, task_type: str, params: dict) -> dict: - enabled = ( - os.environ.get("ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB", "") - .strip() - .lower() - in {"1", "true", "yes", "on"} - ) - task_id = str( - uuid.uuid5(uuid.NAMESPACE_URL, workflow.info().workflow_id) - ) - - if not enabled: - workflow.logger.error( - "TaskExecutorWorkflow refused: stub disabled " - "(set ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB=true only for legacy tests). " - "Real execution belongs in per-repo workers / agent-harness. " - "See docs/task-emission-consumer-contract.md" - ) - raise RuntimeError( - "TaskExecutorWorkflow is disabled (ACTIVITY-WP-0023-T08). " - "Use per-repo executors; do not route production work here." - ) - - workflow.logger.warning( - "TaskExecutorWorkflow stub running (legacy mode)", - extra={"run_id": run_id, "task_type": task_type, "task_id": task_id}, - ) - - await workflow.execute_activity( - persist_task_instance, - { - "id": task_id, - "run_id": run_id, - "type": task_type, - "params": params, - "status": "done", - }, - task_queue=_TASK_QUEUE, - start_to_close_timeout=_ACTIVITY_TIMEOUT, - retry_policy=_RETRY_POLICY, - ) - - return {"task_id": task_id, "status": "done", "legacy_stub": True} diff --git a/tests/rules/test_executor.py b/tests/rules/test_executor.py index cd9c71b..9c04781 100644 --- a/tests/rules/test_executor.py +++ b/tests/rules/test_executor.py @@ -263,7 +263,7 @@ def test_execute_instruction_with_audit_returns_metadata(): assert len(result.prompt_hash) == 64 assert result.model == "test-model" assert result.output_validated is True - assert result.review_required is True + assert result.review_advisory is True def test_execute_instruction_forwards_llm_connect_run_config(): @@ -443,7 +443,7 @@ def test_resilient_report_recovers_valid_prefix_and_quarantines_truncated_tail() result = execute_instruction_with_audit(instr, _Event(), {}, llm) assert result.output_validated is True - assert result.review_required is True + assert result.review_advisory is True assert result.report is not None assert result.report["partial"] is True assert len(result.report["recommendations"]) == 7 @@ -593,7 +593,7 @@ class _MetadataBadLLM: return ("x" * 9000) + "{" -def test_invalid_report_preserves_response_metadata_and_long_preview(): +def test_invalid_report_keeps_metadata_but_discards_provider_output(): llm = _MetadataBadLLM() instr = _instr( id="daily-triage-report", @@ -611,10 +611,11 @@ def test_invalid_report_preserves_response_metadata_and_long_preview(): "usage": {"input_tokens": 1100, "output_tokens": 1200}, } assert result.report["llm_response_metadata"] == result.llm_response_metadata - assert len(result.report["raw_output_preview"]) > 4000 + assert "raw_output_preview" not in result.report + assert "partial_report" not in result.report -def test_execute_instruction_with_audit_preserves_invalid_report_with_sinks( +def test_execute_instruction_with_audit_discards_invalid_report_body( tmp_path, monkeypatch, ): @@ -657,12 +658,13 @@ def test_execute_instruction_with_audit_preserves_invalid_report_with_sinks( assert result.tasks == [] assert result.output_validated is False - assert result.review_required is True + assert result.review_advisory is True assert result.validation_error == "$.recommendations[0]: missing required property 'action'" assert result.report is not None assert result.report["status"] == "validation_failed" - assert result.report["partial_summary"] == "Generated partial triage." - assert result.report["partial_report"] == report_data + assert "partial_summary" not in result.report + assert "partial_report" not in result.report + assert "Generated partial triage." not in str(result.report) assert llm.call_count == 2 @@ -678,7 +680,7 @@ def test_execute_instruction_with_audit_preserves_execution_failure_with_sinks() assert result.tasks == [] assert result.output_validated is False - assert result.review_required is True + assert result.review_advisory is True assert result.validation_error == "LLM_CONNECT_URL is not configured" assert result.report == { "summary": ( @@ -726,8 +728,7 @@ def test_condition_true_calls_llm(): # ── review_required field ───────────────────────────────────────────────────── -def test_review_required_field_on_instruction_def(): - """review_required is a declared field on InstructionDef.""" +def test_legacy_review_required_maps_to_advisory(): defn = InstructionDef( id="test", trusted_fields=["event.attributes.x"], @@ -737,6 +738,7 @@ def test_review_required_field_on_instruction_def(): review_required=True, ) assert defn.review_required is True + assert defn.review_advisory is True def test_instruction_def_accepts_llm_connect_depth_config(): @@ -755,7 +757,7 @@ def test_instruction_def_accepts_llm_connect_depth_config(): assert defn.model_params == {"reasoning_effort": "medium"} -def test_review_required_defaults_to_false(): +def test_review_advisory_defaults_to_false(): defn = InstructionDef( id="test", trusted_fields=[], @@ -763,7 +765,21 @@ def test_review_required_defaults_to_false(): prompt="p", output_schema="schema.json", ) - assert defn.review_required is False + assert defn.review_required is None + assert defn.review_advisory is False + + +def test_conflicting_review_names_are_rejected(): + with pytest.raises(ValueError, match="review_required conflicts"): + InstructionDef( + id="test", + trusted_fields=[], + model="claude-sonnet-4-6", + prompt="p", + output_schema="schema.json", + review_required=True, + review_advisory=False, + ) def test_unknown_root_in_field_path_raises(): diff --git a/tests/test_api_admission.py b/tests/test_api_admission.py new file mode 100644 index 0000000..b416339 --- /dev/null +++ b/tests/test_api_admission.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +import pytest +from fastapi import HTTPException + +from activity_core.api import _validate_context_admission + + +def test_basic_rest_administration_refuses_unknown_shell_query() -> None: + with pytest.raises(HTTPException) as exc_info: + _validate_context_admission( + [{"type": "shell", "query": "run_from_request", "params": {}}] + ) + + assert exc_info.value.status_code == 422 + assert "not registered" in str(exc_info.value.detail) + + +def test_basic_rest_administration_accepts_registered_read_only_query() -> None: + _validate_context_admission( + [ + { + "type": "shell", + "query": "discover_kaizen_projects", + "params": {}, + } + ] + ) diff --git a/tests/test_audit_projection.py b/tests/test_audit_projection.py new file mode 100644 index 0000000..2251cdb --- /dev/null +++ b/tests/test_audit_projection.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from activity_core.audit_projection import bounded_audit_projection + + +def test_audit_projection_drops_raw_and_credential_shaped_fields() -> None: + raw = { + "definition_id": "daily-triage", + "prompt_hash": "a" * 64, + "context": { + "summary": "bounded", + "rendered_prompt": "do not persist", + "messages": [{"content": "do not persist"}], + "provider_response": {"body": "do not persist"}, + "tool_output": "do not persist", + "credential": "do not persist", + "api_key": "do not persist", + "nested": {"password": "do not persist", "safe": True}, + }, + } + + projected = bounded_audit_projection(raw) + + assert projected == { + "definition_id": "daily-triage", + "prompt_hash": "a" * 64, + "context": {"summary": "bounded", "nested": {"safe": True}}, + } + assert "do not persist" not in str(projected) + + +def test_audit_projection_bounds_depth_items_and_strings() -> None: + projected = bounded_audit_projection( + {"text": "x" * 20, "items": list(range(5)), "nested": {"value": 1}}, + max_depth=1, + max_items=2, + max_string=5, + ) + + assert projected == {"text": "xxxxx", "items": ["", ""]} diff --git a/tests/test_bounded_operations.py b/tests/test_bounded_operations.py new file mode 100644 index 0000000..9f4d36e --- /dev/null +++ b/tests/test_bounded_operations.py @@ -0,0 +1,167 @@ +from __future__ import annotations + +import pytest + +from activity_core.bounded_operations import ( + BOUNDED_OPERATION_REGISTRY, + normalize_operation_result, + operation_sources, + validate_bounded_operations, +) + + +def _evidence_sinks(event_type: str) -> list[dict[str, str]]: + return [{"type": "state-hub-progress", "event_type": event_type}] + + +def test_registry_names_only_the_three_accepted_operations() -> None: + assert set(BOUNDED_OPERATION_REGISTRY) == { + "sbom_nexus_ingest", + "forgejo_package_prune", + "cnpg_option_a_backup", + } + + +def test_operation_result_projection_drops_subprocess_output() -> None: + projected = normalize_operation_result( + "cnpg_option_a_backup", + { + "overall": "pass", + "dumped": 3, + "uploaded": 3, + "failed": 0, + "script_exit_code": 0, + "log_tail": "must not persist", + "tool_output": "must not persist", + "credential": "must not persist", + }, + ) + + assert projected == { + "kind": "cnpg_option_a_backup", + "overall": "pass", + "dry_run": False, + "dumped": 3, + "uploaded": 3, + "failed": 0, + "script_exit_code": 0, + } + assert "must not persist" not in str(projected) + + +def test_valid_prune_operation_is_admitted() -> None: + sources = [ + { + "type": "shell", + "query": "forgejo_package_prune", + "operation": "forgejo_package_prune", + "params": { + "prune_script": "/opt/railiance-platform/tools/cmd/forgejo-package-prune", + "live_images_file": "/opt/railiance-platform/live-images.txt", + "apply": True, + "max_versions": 3, + "evidence_sinks": _evidence_sinks("forgejo_package_prune"), + }, + } + ] + + validate_bounded_operations(sources, []) + assert operation_sources(sources)[0][1].temporal_max_attempts == 1 + + +@pytest.mark.parametrize( + ("source", "error"), + [ + ( + {"type": "shell", "query": "run_whatever", "params": {}}, + "not registered", + ), + ( + { + "type": "shell", + "query": "forgejo_package_prune", + "params": {}, + }, + "must declare operation", + ), + ( + { + "type": "shell", + "query": "forgejo_package_prune", + "operation": "forgejo_package_prune", + "params": { + "prune_script": "/tmp/arbitrary-command", + "apply": False, + "max_versions": 3, + "evidence_sinks": _evidence_sinks("forgejo_package_prune"), + }, + }, + "canonical path", + ), + ], +) +def test_shell_admission_fails_closed(source: dict, error: str) -> None: + with pytest.raises(ValueError, match=error): + validate_bounded_operations([source], []) + + +def test_sbom_requires_bounded_limit_and_report_evidence() -> None: + source = { + "type": "sbom-nexus", + "query": "catch_up", + "operation": "sbom_nexus_ingest", + "params": {"apply": True, "limit": 4}, + } + instruction = { + "report_sinks": _evidence_sinks("sbom_catchup"), + } + + with pytest.raises(ValueError, match="limit must be in 1..3"): + validate_bounded_operations([source], [instruction]) + + source["params"]["limit"] = 3 + with pytest.raises(ValueError, match="requires an instruction report"): + validate_bounded_operations([source], []) + + validate_bounded_operations([source], [instruction]) + + +def test_backup_requires_explicit_safe_targets_and_evidence() -> None: + source = { + "type": "shell", + "query": "cnpg_option_a_backup", + "operation": "cnpg_option_a_backup", + "params": { + "backup_script": "/opt/railiance-platform/tools/cmd/cnpg-option-a-backup", + "dry_run": False, + "timeout_seconds": 7200, + "targets": "r01-forgejo-db,r01-state-hub-db", + "evidence_sinks": _evidence_sinks("cnpg_option_a_backup"), + }, + } + + validate_bounded_operations([source], []) + source["params"]["targets"] = "ok,$(unsafe)" + with pytest.raises(ValueError, match="safe names"): + validate_bounded_operations([source], []) + + source["params"]["targets"] = "r01-forgejo-db" + source["params"]["evidence_sinks"] = [{"type": "state-hub-progress"}] + with pytest.raises(ValueError, match="event_type: cnpg_option_a_backup"): + validate_bounded_operations([source], []) + + +def test_definition_cannot_combine_two_bounded_operations() -> None: + prune = { + "type": "shell", + "query": "forgejo_package_prune", + "operation": "forgejo_package_prune", + } + backup = { + "type": "shell", + "query": "cnpg_option_a_backup", + "operation": "cnpg_option_a_backup", + } + + with pytest.raises(ValueError, match="at most one bounded operation"): + validate_bounded_operations([prune, backup], []) diff --git a/tests/test_forgejo_prune_context_resolver.py b/tests/test_forgejo_prune_context_resolver.py index 30a6bff..c8c4b13 100644 --- a/tests/test_forgejo_prune_context_resolver.py +++ b/tests/test_forgejo_prune_context_resolver.py @@ -1,9 +1,11 @@ from __future__ import annotations import json -from unittest import mock +import pytest +from activity_core import activities from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY +from activity_core.context_resolvers.forgejo_prune import forgejo_package_prune class _Completed: @@ -13,7 +15,9 @@ class _Completed: self.returncode = returncode -def test_shell_resolver_runs_forgejo_package_prune(tmp_path, monkeypatch) -> None: +def test_bounded_operation_implementation_runs_forgejo_package_prune( + tmp_path, monkeypatch +) -> None: script = tmp_path / "forgejo-package-prune" script.write_text("#!/usr/bin/env bash\necho '{}'\n", encoding="utf-8") script.chmod(0o755) @@ -29,9 +33,7 @@ def test_shell_resolver_runs_forgejo_package_prune(tmp_path, monkeypatch) -> Non lambda *args, **kwargs: _Completed(payload), ) - result = CONTEXT_RESOLVER_REGISTRY["shell"]().resolve( - "forgejo_package_prune", - None, + result = forgejo_package_prune( {"prune_script": str(script), "max_versions": 3, "apply": False}, ) @@ -46,9 +48,7 @@ def test_apply_without_live_images_file_is_rejected(tmp_path) -> None: script.chmod(0o755) try: - CONTEXT_RESOLVER_REGISTRY["shell"]().resolve( - "forgejo_package_prune", - None, + forgejo_package_prune( {"prune_script": str(script), "max_versions": 3, "apply": True}, ) raise AssertionError("expected RuntimeError") @@ -64,9 +64,7 @@ def test_apply_with_empty_live_images_file_is_rejected(tmp_path) -> None: empty.write_text("", encoding="utf-8") try: - CONTEXT_RESOLVER_REGISTRY["shell"]().resolve( - "forgejo_package_prune", - None, + forgejo_package_prune( { "prune_script": str(script), "apply": True, @@ -75,4 +73,43 @@ def test_apply_with_empty_live_images_file_is_rejected(tmp_path) -> None: ) raise AssertionError("expected RuntimeError") except RuntimeError as exc: - assert "non-empty" in str(exc) or "empty" in str(exc) \ No newline at end of file + assert "non-empty" in str(exc) or "empty" in str(exc) + + +def test_shell_resolver_refuses_mutating_prune_query() -> None: + with pytest.raises(RuntimeError, match="bounded-operation stage"): + CONTEXT_RESOLVER_REGISTRY["shell"]().resolve( + "forgejo_package_prune", None, {} + ) + + +@pytest.mark.asyncio +async def test_execute_bounded_operation_dispatches_prune(monkeypatch) -> None: + monkeypatch.setattr( + "activity_core.context_resolvers.forgejo_prune.forgejo_package_prune", + lambda params: {"kind": "forgejo_package_prune", "deleted_count": 2}, + ) + + result = await activities.execute_bounded_operation( + { + "run_id": "run-1", + "source": { + "type": "shell", + "query": "forgejo_package_prune", + "operation": "forgejo_package_prune", + "params": {}, + "bind_to": "context.prune", + }, + } + ) + + assert result == { + "prune": { + "kind": "forgejo_package_prune", + "apply": False, + "candidate_count": 0, + "deleted_count": 2, + "skipped_protected_count": 0, + "error_count": 0, + } + } diff --git a/tests/test_instruction_evaluation.py b/tests/test_instruction_evaluation.py index 3293257..e10d5c7 100644 --- a/tests/test_instruction_evaluation.py +++ b/tests/test_instruction_evaluation.py @@ -43,7 +43,7 @@ async def test_evaluate_instructions_returns_task_specs_with_audit(monkeypatch) "model": "test-model", "prompt": "Open tasks: {context.summary.open_tasks}", "output_schema": "", - "review_required": False, + "review_advisory": False, } ], "event": {}, @@ -58,7 +58,9 @@ async def test_evaluate_instructions_returns_task_specs_with_audit(monkeypatch) assert spec["source_id"] == "daily-triage" assert spec["model"] == "test-model" assert spec["output_validated"] is True - assert spec["review_required"] is False + assert spec["review_advisory"] is False + assert spec["review_gate_applied"] is False + assert "review_required" not in spec assert spec["prompt_hash"] is not None assert len(spec["prompt_hash"]) == 64 assert result["reports"] == [] @@ -90,7 +92,7 @@ async def test_evaluate_instructions_returns_report_payload(monkeypatch) -> None "model": "test-model", "prompt": "Run report.", "output_schema": "activity-core://schemas/daily-triage-report.json", - "review_required": False, + "review_advisory": False, } ], "event": {}, @@ -145,7 +147,7 @@ async def test_evaluate_instructions_returns_invalid_report_for_report_sinks( "model": "test-model", "prompt": "Run report.", "output_schema": "schemas/daily-triage-report.json", - "review_required": False, + "review_advisory": False, "report_sinks": [{"type": "working-memory", "path": "/tmp"}], } ], @@ -157,10 +159,13 @@ async def test_evaluate_instructions_returns_invalid_report_for_report_sinks( assert len(result["reports"]) == 1 report = result["reports"][0] assert report["output_validated"] is False - assert report["review_required"] is True + assert report["review_advisory"] is True + assert report["review_gate_applied"] is False assert report["validation_error"] == "$.recommendations[0]: missing required property 'wsjf'" assert report["report"]["status"] == "validation_failed" - assert report["report"]["partial_summary"] == "Partial triage." + assert "partial_summary" not in report["report"] + assert "partial_report" not in report["report"] + assert "Partial triage." not in str(report) @pytest.mark.asyncio @@ -210,7 +215,7 @@ async def test_evaluate_instructions_forwards_llm_connect_depth_config(monkeypat "model_params": {"reasoning_effort": "medium"}, "prompt": "Run report.", "output_schema": "schemas/daily-triage-report.json", - "review_required": False, + "review_advisory": False, } ], "event": {}, diff --git a/tests/test_ops_evidence_sinks.py b/tests/test_ops_evidence_sinks.py index 96e5a95..ac04a8b 100644 --- a/tests/test_ops_evidence_sinks.py +++ b/tests/test_ops_evidence_sinks.py @@ -225,6 +225,54 @@ def test_state_hub_progress_prefers_workplan_id(monkeypatch) -> None: assert "workstream_id" not in body +def test_bounded_operation_progress_drops_raw_and_credential_fields(monkeypatch) -> None: + posts: list[dict[str, Any]] = [] + + monkeypatch.setattr(httpx, "get", lambda *args, **kwargs: DummyResponse([])) + + def fake_post(url: str, **kwargs: Any) -> DummyResponse: + posts.append({"url": url, **kwargs}) + return DummyResponse({"id": "progress-prune-1"}) + + monkeypatch.setattr(httpx, "post", fake_post) + persist_ops_inventory_evidence( + { + "activity_id": "weekly-forgejo-package-prune", + "run_id": "12345678-aaaa-bbbb-cccc-123456789abc", + "context_sources": [ + { + "type": "shell", + "query": "forgejo_package_prune", + "bind_to": "context.prune", + "params": { + "evidence_sinks": [ + { + "type": "state-hub-progress", + "state_hub_url": "http://state-hub.test", + "event_type": "forgejo_package_prune", + } + ] + }, + } + ], + "context": { + "prune": { + "kind": "forgejo_package_prune", + "apply": True, + "deleted_count": 2, + "tool_output": "must not persist", + "credential": "must not persist", + "provider_response": {"raw": "must not persist"}, + } + }, + } + ) + + body = posts[0]["json"] + assert body["detail"]["probe"]["deleted_count"] == 2 + assert "must not persist" not in str(body) + + def test_core_hub_stabilization_sink_posts_progress(monkeypatch) -> None: posts: list[dict[str, Any]] = [] diff --git a/tests/test_report_sinks.py b/tests/test_report_sinks.py index 97d7f2a..c1cc2b6 100644 --- a/tests/test_report_sinks.py +++ b/tests/test_report_sinks.py @@ -36,7 +36,8 @@ def _payload(sinks: list[dict[str, Any]]) -> dict[str, Any]: "prompt_hash": "abc123", "model": "test-model", "output_validated": True, - "review_required": False, + "review_advisory": False, + "review_gate_applied": False, "validation_error": None, "llm_response_metadata": { "finish_reason": "stop", @@ -66,7 +67,9 @@ def test_working_memory_sink_writes_idempotently(tmp_path) -> None: text = note.read_text(encoding="utf-8") assert "activity_core_run_id: 12345678-aaaa-bbbb-cccc-123456789abc" in text assert "output_validated: true" in text - assert "review_required: false" in text + assert "review_advisory: false" in text + assert "review_gate_applied: false" in text + assert "review_required:" not in text assert "model: test-model" in text assert "LLM response metadata:" in text assert '"finish_reason": "stop"' in text @@ -121,7 +124,8 @@ def test_state_hub_progress_sink_posts(monkeypatch) -> None: assert "workstream_id" not in posts[0]["json"] assert posts[0]["json"]["detail"]["activity_core_run_id"] == payload_run_id() assert posts[0]["json"]["detail"]["output_validated"] is True - assert posts[0]["json"]["detail"]["review_required"] is False + assert posts[0]["json"]["detail"]["review_advisory"] is False + assert posts[0]["json"]["detail"]["review_gate_applied"] is False assert posts[0]["json"]["detail"]["llm_response_metadata"] == { "finish_reason": "stop", "usage": {"output_tokens": 50}, diff --git a/tests/test_resolve_context_binding.py b/tests/test_resolve_context_binding.py index 124050e..0931634 100644 --- a/tests/test_resolve_context_binding.py +++ b/tests/test_resolve_context_binding.py @@ -158,3 +158,37 @@ async def test_event_payload_context_fails_when_required_envelope_missing() -> N } ], ) + + +@pytest.mark.asyncio +async def test_resolve_context_does_not_execute_bounded_shell_operation() -> None: + snapshot = await resolve_context( + [ + { + "type": "shell", + "query": "forgejo_package_prune", + "operation": "forgejo_package_prune", + "params": {}, + "bind_to": "context.prune", + } + ] + ) + + assert snapshot == { + "prune": {"operation": "forgejo_package_prune", "status": "pending"} + } + + +@pytest.mark.asyncio +async def test_resolve_context_refuses_unknown_shell_query() -> None: + with pytest.raises(ApplicationError, match="not registered as read-only"): + await resolve_context( + [ + { + "type": "shell", + "query": "arbitrary_command", + "params": {}, + "bind_to": "context.bad", + } + ] + ) diff --git a/tests/test_run_artifacts.py b/tests/test_run_artifacts.py index a518d40..89f0c8e 100644 --- a/tests/test_run_artifacts.py +++ b/tests/test_run_artifacts.py @@ -11,6 +11,7 @@ from activity_core.run_artifacts import ( artifacts_from_ops_result, build_forgejo_blob_url, match_ops_runs_to_activity_run, + public_context_keys, ) @@ -39,6 +40,17 @@ def test_build_forgejo_blob_url_rejects_traversal() -> None: ) +def test_public_context_keys_hide_credential_and_raw_payload_names() -> None: + assert public_context_keys( + { + "safe_summary": {}, + "credential": "must-drop", + "tool_output": "must-drop", + "provider_response": {}, + } + ) == ["safe_summary"] + + def test_artifacts_from_ops_result() -> None: arts = artifacts_from_ops_result( { diff --git a/tests/test_sbom_nexus_context_resolver.py b/tests/test_sbom_nexus_context_resolver.py index 88ecd71..9db7297 100644 --- a/tests/test_sbom_nexus_context_resolver.py +++ b/tests/test_sbom_nexus_context_resolver.py @@ -616,7 +616,7 @@ def _instruction(): "max_tokens": 1, "prompt": "Deterministic SBOM catch-up report.", "output_schema": "", - "review_required": False, + "review_advisory": False, "report_sinks": [ { "type": "state-hub-progress", diff --git a/workplans/ACTIVITY-WP-0035-intent-boundary-guardrails.md b/workplans/ACTIVITY-WP-0035-intent-boundary-guardrails.md index 53df8ff..eb2e4e4 100644 --- a/workplans/ACTIVITY-WP-0035-intent-boundary-guardrails.md +++ b/workplans/ACTIVITY-WP-0035-intent-boundary-guardrails.md @@ -4,7 +4,7 @@ type: workplan title: "Make the execution boundary enforceable and the review contract truthful" domain: infotech repo: activity-core -status: ready +status: active owner: codex topic_slug: activity-core priority: high @@ -92,7 +92,7 @@ and enforce it in definition admission and workflow structure. At completion: ```task id: ACTIVITY-WP-0035-T01 -status: todo +status: done priority: high ``` @@ -117,7 +117,7 @@ without making activity-core a general executor. ```task id: ACTIVITY-WP-0035-T02 -status: wait +status: done priority: high ``` @@ -145,7 +145,7 @@ query without an explicit reviewed registry entry and tests prove each refusal. ```task id: ACTIVITY-WP-0035-T03 -status: wait +status: done priority: high ``` @@ -170,7 +170,7 @@ operations pass unit, retry, redaction, and workflow-order tests. ```task id: ACTIVITY-WP-0035-T04 -status: todo +status: done priority: medium ``` @@ -192,7 +192,7 @@ audit claim. ```task id: ACTIVITY-WP-0035-T05 -status: todo +status: done priority: medium ``` @@ -217,7 +217,7 @@ migration path are recorded in ADR-003 or a successor ADR. ```task id: ACTIVITY-WP-0035-T06 -status: wait +status: done priority: medium ``` @@ -238,7 +238,7 @@ not close the live acceptance item. ```task id: ACTIVITY-WP-0035-T07 -status: todo +status: done priority: low ``` @@ -262,7 +262,7 @@ activity-core is a general task executor. ```task id: ACTIVITY-WP-0035-T08 -status: wait +status: progress priority: high ``` @@ -281,15 +281,32 @@ Depends on T02–T07 as applicable. ## Acceptance -- [ ] `INTENT.md` explicitly distinguishes work-item lifecycle from ops-run delivery state -- [ ] The bounded-operation exception is governed by a reviewed contract, not convention -- [ ] Definition sync refuses unknown or incomplete mutating operation declarations -- [ ] Context resolvers are read-only; mutations run in an explicit bounded-operation phase -- [ ] Instruction audit wording matches persisted evidence and raw sensitive payloads remain excluded -- [ ] `review_required` has truthful, tested hold/release or advisory-only semantics -- [ ] Misleading executor/CRUD/README surfaces are removed or precisely qualified +- [x] `INTENT.md` explicitly distinguishes work-item lifecycle from ops-run delivery state +- [x] The bounded-operation exception is governed by a reviewed contract, not convention +- [x] Definition sync refuses unknown or incomplete mutating operation declarations +- [x] Context resolvers are read-only; mutations run in an explicit bounded-operation phase +- [x] Instruction audit wording matches persisted evidence and raw sensitive payloads remain excluded +- [x] `review_required` has truthful, tested hold/release or advisory-only semantics +- [x] Misleading executor/CRUD/README surfaces are removed or precisely qualified - [ ] Full tests pass and production rollout preserves current schedules and evidence +## Implementation evidence — 2026-08-23 + +- ACT-ADR-007 accepted and recorded as State Hub decision + `2b1f0c01-1d6f-4fdc-b3e7-537ce0a0a1f8`. +- Strict parsing admitted all 12 checked-in definitions; refusal tests cover + unknown shell queries, missing operation declarations, arbitrary script + paths, missing evidence, target/limit bounds, and multiple mutations. +- Full suite: `453 passed, 1 skipped` in 139.62 seconds; subsequent API + admission/render-focused suite: `40 passed`. +- PostgreSQL migration `0008 → 0009 → 0008 → 0009` passed in an isolated + PostgreSQL 16 database. The final state has no `task_instances` table and has + `task_spawn_log.review_advisory`. +- Production removal inventory before migration: zero `task_instances`, no + worker feature flag, and zero Temporal `TaskExecutorWorkflow` executions. +- Compatibility detail is retained in + `history/2026-08-23-bounded-operation-compatibility-report.md`. + ## Gap disposition | 2026-08-23 gap | This workplan |