diff --git a/.custodian-brief.md b/.custodian-brief.md deleted file mode 100644 index 0baac29..0000000 --- a/.custodian-brief.md +++ /dev/null @@ -1,75 +0,0 @@ - -# Custodian Brief — activity-core - -**Domain:** infotech -**Last synced:** 2026-10-13 19:10 UTC -**State Hub:** http://127.0.0.1:8000 *(adjust if running on a remote machine)* - -## Active Workstreams - -### Adopt the daily security-flow schedule (CUST-WP-0079-T03) -Progress: 1/3 done | workplan_id: `ee97869b-dca2-54c9-aa18-0b479c244fb3` - -**Open tasks:** -- ! Activation `d0d5fddd` -- ! First unassisted fire `d692a3eb` - -### Repository grant v2: publish scope for governed pushes (REINAH-WP-0003-T04) -Progress: 1/3 done | workplan_id: `b6b5d39e-cd56-5a91-a67e-dcdf7b8d147c` - -**Open tasks:** -- ! rein-aharness contract confirmation `78c39857` -- ! assembly-checkpoint definitions adopt v2 `0c3c628b` - -### Adopt the Glas profile-driven execution contract -Progress: 4/5 done | workplan_id: `256dad13-28b4-5361-ab8a-7d1373a5d14b` - -**Open tasks:** -- ! Prove on one definition `e71423f9` - *(wait: No admitted Glas profile: waits on GLAS-WP-0012 (upstream GLAS-IN-0002, sand-boxer nsenter denial); do not rerun the old pilot.)* - -### Remove recurring deployment exceptions through governed GitOps adoption -Progress: 1/3 done | workplan_id: `12798750-3eee-5d81-a411-c3c4c5ce2d2c` - -**Open tasks:** -- ! Adopt the declared resource set into railiance01 ArgoCD `af09865d` - *(wait: Automated sync after RPF-WP-0048-T02 soak; founder confirmation given by D2.)* -- ! Establish standing authority for bounded routine releases `8e2b4ed9` - *(wait: RPF-IN-0011: return dedicated release principal/custody owner task and native acceptance; D2 scope already accepted)* - -### Owner-return pickup adapter for State Hub STATE-WP-0094-T03 -Progress: 5/6 done | workplan_id: `e65f59d4-2272-557d-9071-734c88186d19` - -**Open tasks:** -- ! Founder-reviewed activation `254526ce` - *(wait: Founder accepted memo 2 (approval 77993081); trials wait for the first live C-25 owner return)* - -### Repo Manager bounded freshness ActivityDefinition (RMGR-WP-0020-T03) -Progress: 2/3 done | workplan_id: `85e9d104-05e7-54d3-86d2-466af8cc98c4` - -**Open tasks:** -- ! Founder-reviewed activation `9aeaf03d` - *(wait: Activation follows RMGR-WP-0020-T04 and is a founder action)* - -### Scheduling role for recurring platform backup/restore evidence runs -Progress: 5/8 done | workplan_id: `e1bfb5a4-e647-592d-883f-fe8728ad2f5f` - -**Open tasks:** -- ! Admit a bounded operation and credential lane `250c6e8e` - *(wait: Offline contract is implemented; operation admission, credential scope and live proof wait on platform acceptance in T05.)* -- ! Agree the platform runner contract `a2d95c80` - *(wait: Platform owner must accept the candidate request/receipt contract, execution authority checks, limits and credential scope before runtime wiring.)* -- ! Refresh the existing backup consumer after the owner repairs the lane `020d19be` - *(wait: KV4 provider repair is delivered; repo-manager historical-prefix acceptance is delivered (receipt 39f2c067, 2026-10-06). Separate approved CNPG hostPath/Forgejo GitOps bundle delivery, ESO/env refresh and native consumer receipt remain. Founder per-change go is required before live promotion.)* - -## Inbox Hygiene - -**Missing thread_id:** 1 unread message(s) lack supersession chains. - ---- -## MCP Orientation (when available) - -If the state-hub MCP server is reachable, call: -`get_domain_summary("infotech")` -This provides richer cross-domain context. -If the MCP call fails, use this file as your orientation source. diff --git a/.gitignore b/.gitignore index 36b13f1..887a23a 100644 --- a/.gitignore +++ b/.gitignore @@ -174,3 +174,5 @@ cython_debug/ # PyPI configuration file .pypirc +# Local owner-generated orientation snapshot. +/.custodian-brief.md diff --git a/SCOPE.md b/SCOPE.md index 78f9266..9fa9ac3 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -368,3 +368,16 @@ Unstaged source edits are preserved; unstaged generated-file edits and unresolve conflicts stop the commit. The existing generated-only guard and secret scan remain in place. This adoption covers WORK-RECORDS.md only; other owner indexes retain their own generators. It makes no schedule or public-release change. + + +## Custodian deterministic operations + +The existing bounded-operation stage registers custodian_security_flow and +custodian_allocation_review: fixed owner commands with trusted operator-bound +source/fleet/private-state inputs, explicit definition parameters, timeout +containment and compact counted/hash receipts. Both definitions remain disabled. +No coding-agent task or green-commit profile executes these routines. Runtime +placement, accepted mounts/network, delivery and first unattended proof remain +unverified; see docs/custodian-bounded-operations.md. This adds no scheduler or +generic shell. The generated .custodian-brief.md is now an ignored local +orientation snapshot; its existing bytes remain local. diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 9a82870..81d58ac 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -271,6 +271,7 @@ | task | ACTIVITY-WP-0047-T01 | done | — | workplans/ACTIVITY-WP-0047-daily-security-flow-adoption.md | | task | ACTIVITY-WP-0047-T02 | wait | — | workplans/ACTIVITY-WP-0047-daily-security-flow-adoption.md | | task | ACTIVITY-WP-0047-T03 | wait | — | workplans/ACTIVITY-WP-0047-daily-security-flow-adoption.md | +| task | ACTIVITY-WP-0047-T04 | done | — | workplans/ACTIVITY-WP-0047-daily-security-flow-adoption.md | | task | ACTIVITY-WP-ADHOC-2026-06-01-T01 | done | — | workplans/ADHOC-2026-06-01.md | | task | ACTIVITY-WP-ADHOC-2026-06-01-T02 | done | — | workplans/ADHOC-2026-06-01.md | | task | ACTIVITY-WP-ADHOC-2026-06-01-T03 | done | — | workplans/ADHOC-2026-06-01.md | diff --git a/activity-definitions/daily-security-flow.md b/activity-definitions/daily-security-flow.md index b45014e..b555ea2 100644 --- a/activity-definitions/daily-security-flow.md +++ b/activity-definitions/daily-security-flow.md @@ -10,48 +10,46 @@ trigger: cron_expression: "30 7 * * *" timezone: Europe/Berlin misfire_policy: skip +context_sources: + - type: shell + query: custodian_security_flow + operation: custodian_security_flow + required: true + params: + apply: true + timeout_seconds: 240 + evidence_sinks: + - type: state-hub-progress + event_type: custodian_security_flow + author: activity-core + bind_to: context.custodian_result --- # Daily Security Flow -Adopted from the-custodian CUST-WP-0079-T03, tracked in ACTIVITY-WP-0047. -Activity Core owns the schedule; the-custodian owns the runner -(`scripts/security_flow_daily.py`), its report and its delivery checkpoints. +Disabled source preparation using the existing ACT-ADR-007 bounded-operation +stage. This definition emits no agent tasks or ops_runs. The worker runs one +code-fixed owner command, retains a counted/hash receipt in the activity context, +and passes it through the existing progress evidence sink. No model or coding +profile executes this routine. No arbitrary command, import, executable or path +is accepted from definition parameters. -Each fire emits one ops_run for a rein on the workstation profile, which runs a -fixed command in the-custodian checkout: +The operator must bind the actual deployed owner code, fleet snapshot root, +private external state, and scoped Hub access/input request before enabling. +See `docs/custodian-bounded-operations.md`. These bindings are absent by default; +missing paths, redirected paths and unbound requests fail before subprocess +execution. This is not a credential-isolation claim or a runtime acceptance. -```text -python3 scripts/security_flow_daily.py --apply -``` +Single Temporal attempt; skip misfires. Owner checkpoint/retention semantics +govern retries. A completed command is not owner acknowledgment or independent +missed-run recovery. No schedule has been activated or deployed. Rollback is +`enabled: false`, retaining reports/checkpoints. -The runner resolves hub decisions already recorded as resolved in repo decision -logs (the sanctioned `POST /decisions/{id}/resolve`), writes a deterministic -JSON report, and messages coding agents only when READY/UNBLOCKED change and the -founder only when ATTENDED is nonempty. Its per-recipient checkpoint and run -lock make a repeated or overlapping fire safe. - -- Daily 07:30 Europe/Berlin; misfires are skipped, never replayed. -- No repository grant: the run reads repo files and changes nothing in any - repository. A rein must not commit or push for this run. -- Activity Core does not run the script itself (out of scope: general - execution); it dispatches and records the ops_run outcome. - -**Not yet enabled.** Activation (ACTIVITY-WP-0047-T02) flips `enabled: true` -once the claiming rein confirms it can run the command against workstation -checkouts. Rollback: `enabled: false`. - -```rule -id: run-daily-security-flow -condition: "True" -action: - task_template: Run the daily security flow - description: 'In the-custodian checkout run exactly: python3 scripts/security_flow_daily.py --apply. Report its exit code and the printed JSON report. Do not edit, commit or push any repository and do not send messages yourself; the script handles delivery.' - target_repo: the-custodian - priority: medium - labels: ["automated", "security-flow"] - harness_profile_ref: harness.agent-dev-local@1.1.1 - execution_refs: - correlation_id: CUST-WP-0079-T03 - assignment_ref: ACTIVITY-WP-0047-T01 -``` +ACTIVITY-WP-0047 / CUST-WP-0079-T03 owns acceptance. The former +`harness.agent-dev-local@1.1.1` rule was incompatible: its green-commit-only +allowlist cannot invoke Python and its sandbox denies Hub egress. The fixed +owner command is `scripts/security_flow_daily.py --apply`, with operator-bound +`--root`, `--hub` and `--state-dir`. The owner script performs decision sync and +deduplicated messages; the adapter does not message independently. Its receipt +does not prove transport delivery or acknowledgment. The old ops_run-based +first-fire criterion is replaced with bounded activity/report/sink receipts. diff --git a/activity-definitions/weekly-railiance-allocation-review.md b/activity-definitions/weekly-railiance-allocation-review.md index efb9ced..6bd7445 100644 --- a/activity-definitions/weekly-railiance-allocation-review.md +++ b/activity-definitions/weekly-railiance-allocation-review.md @@ -10,59 +10,45 @@ trigger: cron_expression: "0 8 * * 1" timezone: Europe/Berlin misfire_policy: skip -context_sources: [] +context_sources: + - type: shell + query: custodian_allocation_review + operation: custodian_allocation_review + required: true + params: + timeout_seconds: 240 + evidence_sinks: + - type: state-hub-progress + event_type: custodian_allocation_review + author: activity-core + bind_to: context.custodian_result --- # Weekly Railiance allocation review -CUST-WP-0071-T05 source preparation. Activity-core owns durable scheduling and -execution receipts; the-custodian owns the deterministic retained evidence join -in `scripts/weekly_allocation_report.py`. Collector inputs remain owned by -railiance-cluster, rapp-telemetry, workload owners and resource-control. +Disabled source preparation using the existing ACT-ADR-007 bounded-operation +stage. This definition emits no agent tasks or ops_runs. The worker runs one +code-fixed owner command, retains a counted/hash receipt in the activity context, +and passes it through the existing progress evidence sink. No model or coding +profile executes this routine. No arbitrary command, import, executable or path +is accepted from definition parameters. -**Disabled and not ready for registration or activation.** No durable Railiance -executor/profile, input capture path, retention location, delivery channel or -review-owner acknowledgment is invented here. The request template contains null -bindings and the command refuses execution until they are supplied. A workstation -rein is not the accepted durable Railiance executor required by this workplan. -Bind the accepted profile before projection into any profile-enforcing runtime. +The operator must bind the actual deployed owner code, fleet snapshot root, +private external state, and scoped Hub access/input request before enabling. +See `docs/custodian-bounded-operations.md`. These bindings are absent by default; +missing paths, redirected paths and unbound requests fail before subprocess +execution. This is not a credential-isolation claim or a runtime acceptance. -The fixed command, in the-custodian checkout, is: +Single Temporal attempt; skip misfires. Owner checkpoint/retention semantics +govern retries. A completed command is not owner acknowledgment or independent +missed-run recovery. No schedule has been activated or deployed. Rollback is +`enabled: false`, retaining reports/checkpoints. -```text -python3 scripts/weekly_allocation_report.py --request config/weekly-allocation-request.json -``` - -An admitted executor must provide a per-run stable request binding the actual -seven-day query captures, matching cluster/allocation snapshot and provenance, -portfolio input, previous retained report, explicit evaluation timestamp and -durable output path. Retrying the identical request is idempotent; changed input -at an existing report path is refused. Never replace a failed capture with old -data carrying a new timestamp. The runner reports missing or stale data as gaps. - -Use recommendation IDs as upsert keys into existing owner work records after -accepted routing. No task fan-out, sizing change or delivery is performed by this -script. The task below asks an executor to run one fixed command, not to devise -an allocation or start a coding session. An ops_run success alone is not an owner -acknowledgment, independent missed-run proof or useful-operation acceptance. - -Enable only after CUST-WP-0071-T04 and executor-owner acceptance, including scoped -read access, retention beyond seven days, report delivery/acknowledgment and an -outside-runner stale/failed/missed-run recovery route. Exercise retry and missed -run handling; retain the first unattended scheduled report and owner receipt. -Rollback is `enabled: false`; reports are retained. Misfires are skipped, so -explicit missed-run detection and recovery are required before enablement. - -```rule -id: run-weekly-railiance-allocation-review -condition: "True" -action: - task_template: Retain the weekly Railiance allocation assessment - description: 'In the-custodian checkout run exactly: python3 scripts/weekly_allocation_report.py --request config/weekly-allocation-request.json. Report the exit code and printed JSON receipt. Use only the accepted executor-owned input request; do not collect credentials, invent inputs, resize workloads, edit or commit source, create owner tasks or send messages.' - target_repo: the-custodian - priority: medium - labels: ["automated", "allocation-review", "bounded-operation"] - execution_refs: - correlation_id: CUST-WP-0071-T05 - assignment_ref: CUST-WP-0071-T05 -``` +CUST-WP-0071-T05 owns acceptance, after T04. The fixed owner command is +`scripts/weekly_allocation_report.py --request `. +The request binds real retained captures, explicit as_of and durable output; +inputs remain under accepted roots and output stays in private runtime state. +The committed null request template is not production input. Stable recommendation +IDs are downstream upsert keys; the adapter neither creates owner work records +nor delivers their acknowledgments. Actual collection, durable retention, +accepted owner routing, weekly report receipt and missed-run recovery remain gates. diff --git a/docs/custodian-bounded-operations.md b/docs/custodian-bounded-operations.md new file mode 100644 index 0000000..c72485a --- /dev/null +++ b/docs/custodian-bounded-operations.md @@ -0,0 +1,73 @@ +# Custodian bounded operations + +Daily security-flow and weekly allocation definitions remain disabled. They use +the existing ACT-ADR-007 bounded-operation stage and Temporal schedule ownership, +with one fixed owner command per run. They emit no agent task, ops_run or model +invocation. The former harness.agent-dev-local@1.1.1 pin is incompatible: its +green-commit-only tools forbid Python, its sandbox denies Hub egress, and its +operational readiness is unverified. + +## Trusted operator bindings + +| Environment variable | Required contract | +| --- | --- | +| ACTIVITY_CORE_CUSTODIAN_ROOT | Absolute pinned owner source bundle, including owner scripts and dependencies. | +| ACTIVITY_CORE_CUSTODIAN_FLEET_ROOT | Absolute accepted repository snapshot/mount root, without credential directories. | +| ACTIVITY_CORE_CUSTODIAN_STATE_ROOT | Precreated private directory, worker-owned, mode 0700, outside source and fleet roots; retained/backed up by executor owner. | +| ACTIVITY_CORE_CUSTODIAN_REPOS | Daily flow's explicit comma-separated canonical repo slugs, each with workplans under fleet root. | +| ACTIVITY_CORE_CUSTODIAN_HUB | Daily flow's explicitly admitted HTTP(S) base URL without embedded credentials, query or fragment. | +| ACTIVITY_CORE_CUSTODIAN_ALLOCATION_REQUEST | Weekly executor-owned request JSON within private state, binding real inputs, fixed evaluation time and retained output. | + +These are trusted deployment configuration, never definition command parameters. +Missing bindings, traversal, symlinks, unbound inputs and output/input aliases +are refused before execution. Weekly inputs remain under accepted roots and +output stays in private state. The state root must be separate: when fleet root +is /home/worsch, its ~/.local/state default is not accepted. Provision a separate +retained mount and precreate its security-flow subdirectory. + +Keep paths/request files protected from concurrent untrusted writers and mount +source read-only. Checks are not a sandbox or credential-isolation boundary. +Owner code and dependencies need accepted revision/digest. The adapter uses the +worker Python and installs nothing. Daily decision sync scans decision logs +throughout the fleet root; reporting scans the explicit repository list. Bind +the root to approved decision-sync scope, not an arbitrary home directory. + +## Execution and receipts + +Daily: worker Python runs security_flow_daily.py with fixed --root, --repos, +--hub, --state-dir, and --apply only when explicitly declared. Owner locking +and checkpoints govern duplicate messages. The adapter suppresses stdout/stderr, +requires a refreshed valid report.json, and retains only SHA-256, exit status, +apply flag and category counts. No message bodies enter its receipt. + +Weekly: worker Python runs weekly_allocation_report.py --request with the bound +request. The owner's immutable retention and identical retry semantics remain, +including unchanged mtime on identical retries. The adapter executes a private snapshot of validated request bytes, preserving +relative-path resolution; pointer changes cannot redirect that run. +Freeze each request across retries; use a new retained path for revised evidence. The receipt contains +report hash and recommendation count, without inventing measurements. + +Both use fixed argv, no shell, at most 240 seconds within a 300-second Temporal +activity timeout, and one attempt. Subprocess timeout terminates the process +group including helpers. Nonzero exits and missing/invalid reports fail; raw +command errors are not persisted. Checkpointed writes can precede failure: +inspect owner state before retry. No parallel scheduler/retry loop is added. + +The existing workflow retains the compact result in activity context and sends +it through the declared State Hub progress evidence sink. This replaces ops_run +close evidence for these routines. Command success does not prove sink delivery +or owner acknowledgment; verify sink outcomes and owner artifacts separately. + +## Remaining acceptance + +ACTIVITY-WP-0047-T02/T03 retain daily activation/first fire; CUST-WP-0071-T05 +retains weekly acceptance after T04. Runtime owner must verify placement, scoped +mounts, Python/dependencies, retained state/backup, admitted Hub connectivity +and exact owner command behavior. No bindings or schedule installation are +established by source tests. Exercise stale/missed/failed-run detection, +ambiguous failure recovery, delivery and owner acknowledgment. Misfires skip; +recovery requires a named owner. Weekly collection and recommendation upserts +also remain. Retain the first unattended activity, report, sink and owner receipts. + +tests/test_custodian_operations.py uses temporary fixture commands or mocked +dispatch. It never invokes the real security command, Hub or schedule activation. diff --git a/src/activity_core/activities.py b/src/activity_core/activities.py index 5e07ef7..90f3108 100644 --- a/src/activity_core/activities.py +++ b/src/activity_core/activities.py @@ -11,6 +11,7 @@ activities that need DB access. from __future__ import annotations +import asyncio import json import uuid from datetime import datetime, timezone @@ -246,7 +247,11 @@ async def execute_bounded_operation(payload: dict[str, Any]) -> dict[str, Any]: 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": + if spec.operation_id in {"custodian_security_flow", "custodian_allocation_review"}: + from activity_core.custodian_operations import run_custodian_operation + + result = await asyncio.to_thread(run_custodian_operation, spec.operation_id, params) + elif spec.operation_id == "forgejo_package_prune": from activity_core.context_resolvers.forgejo_prune import forgejo_package_prune result = forgejo_package_prune(params) diff --git a/src/activity_core/bounded_operations.py b/src/activity_core/bounded_operations.py index 960f204..1df0cf7 100644 --- a/src/activity_core/bounded_operations.py +++ b/src/activity_core/bounded_operations.py @@ -27,6 +27,20 @@ class BoundedOperationSpec: _SPECS = ( + BoundedOperationSpec( + operation_id="custodian_security_flow", source_type="shell", query="custodian_security_flow", + resolve_before_execute=False, max_timeout_seconds=300, temporal_max_attempts=1, + idempotency="single Temporal attempt; owner report/checkpoint retry contract", + credential_route="none; trusted operator binds scoped owner runtime and Hub access", + evidence_mode="context-evidence-sink", + ), + BoundedOperationSpec( + operation_id="custodian_allocation_review", source_type="shell", query="custodian_allocation_review", + resolve_before_execute=False, max_timeout_seconds=300, temporal_max_attempts=1, + idempotency="single Temporal attempt; owner report/checkpoint retry contract", + credential_route="none; trusted operator binds scoped owner runtime and Hub access", + evidence_mode="context-evidence-sink", + ), BoundedOperationSpec( operation_id="sbom_nexus_ingest", source_type="sbom-nexus", @@ -139,6 +153,9 @@ def pending_operation_value(spec: BoundedOperationSpec) -> dict[str, str]: 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 in {"custodian_security_flow", "custodian_allocation_review"}: + keys = ("kind", "status", "exit_code", "report_sha256", "counts", "apply", "owner_acknowledgment") + return {key: result.get(key) for key in keys} if operation_id == "forgejo_package_prune": return { "kind": operation_id, @@ -248,7 +265,11 @@ def validate_bounded_operations( raise ValueError(f"bounded operation {spec.operation_id} params must be a mapping") _validate_timeout(spec, params) - if spec.operation_id == "sbom_nexus_ingest": + if spec.operation_id in {"custodian_security_flow", "custodian_allocation_review"}: + from activity_core.custodian_operations import validate_params + validate_params(spec.operation_id, params) + _require_context_evidence(params, spec.operation_id) + elif spec.operation_id == "sbom_nexus_ingest": _validate_sbom(params, instructions) elif spec.operation_id == "forgejo_package_prune": _validate_prune(params) diff --git a/src/activity_core/custodian_operations.py b/src/activity_core/custodian_operations.py new file mode 100644 index 0000000..de7a67c --- /dev/null +++ b/src/activity_core/custodian_operations.py @@ -0,0 +1,177 @@ +"""Fixed Custodian owner commands for ACT-ADR-007; no definition-supplied argv. + +Operator environment supplies trusted deployment bindings. These checks prevent +accidental redirection; they are not a credential-isolation sandbox. +""" +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path +import re +import signal +import stat +import subprocess +import sys +import time +import tempfile +from urllib.parse import urlsplit + +OPERATIONS = {"custodian_security_flow", "custodian_allocation_review"} +_MAX_JSON_BYTES = 8 * 1024 * 1024 + + +def validate_params(operation: str, params: dict) -> None: + if operation not in OPERATIONS: + raise ValueError("unknown Custodian operation") + allowed = {"timeout_seconds", "evidence_sinks"} + if operation == "custodian_security_flow": + allowed.add("apply") + if type(params.get("apply")) is not bool: + raise ValueError("security flow requires explicit boolean apply") + if set(params) - allowed: + raise ValueError("Custodian operation rejects command/path/unknown definition parameters") + timeout = params.get("timeout_seconds", 240) + if type(timeout) is not int or not 1 <= timeout <= 240: + raise ValueError("Custodian operation timeout must be an integer in 1..240") + + +def _path(value: str | None, *, directory=False, existing=True) -> Path: + if not value: + raise ValueError("Custodian runtime binding missing") + path = Path(value) + if not path.is_absolute() or ".." in path.parts: + raise ValueError("Custodian runtime paths must be absolute without traversal") + if any(part.is_symlink() for part in [path, *path.parents]): + raise ValueError("Custodian runtime paths must not traverse symlinks") + if existing and (not path.is_dir() if directory else not path.is_file()): + raise ValueError("Custodian runtime path unavailable") + return path + + +def _json(path: Path) -> tuple[dict, bytes]: + if path.stat().st_size > _MAX_JSON_BYTES: + raise ValueError("Custodian evidence exceeds size bound") + raw = path.read_bytes() + value = json.loads(raw) + if not isinstance(value, dict): + raise ValueError("Custodian evidence must be an object") + return value, raw + + +def _private_state() -> Path: + path = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_STATE_ROOT"), directory=True) + info = path.stat() + if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o700: + raise ValueError("Custodian state root must be owned by executor with mode 0700") + return path + + +def _within(path: Path, roots: list[Path]) -> bool: + return any(path == root or root in path.parents for root in roots) + + +def build_command(operation: str, params: dict) -> tuple[list[str], Path, Path, dict | None]: + validate_params(operation, params) + root = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_ROOT"), directory=True) + fleet = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_FLEET_ROOT"), directory=True) + state = _private_state() + if _within(state, [root, fleet]): + raise ValueError("Custodian runtime state must be outside source/fleet checkouts") + script_name = "security_flow_daily.py" if operation == "custodian_security_flow" else "weekly_allocation_report.py" + script = _path(str(root / "scripts" / script_name)) + argv = [sys.executable, str(script)] + request = None + if operation == "custodian_security_flow": + repos = os.environ.get("ACTIVITY_CORE_CUSTODIAN_REPOS", "").split(",") + if not repos or any(re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", repo) is None for repo in repos): + raise ValueError("explicit canonical security-flow repository list required") + for repo in repos: + _path(str(fleet / repo / "workplans"), directory=True) + hub = os.environ.get("ACTIVITY_CORE_CUSTODIAN_HUB", "") + parsed = urlsplit(hub) + if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}: + raise ValueError("explicit non-credential Custodian Hub base URL required") + state = _path(str(state / "security-flow"), directory=True) + for name in ("run.lock", "report.json", "delivery.json", "report.json.tmp", "delivery.json.tmp"): + _path(str(state / name), existing=False) + argv += ["--root", str(fleet), "--repos", ",".join(repos), "--hub", hub, "--state-dir", str(state)] + if params["apply"]: + argv.append("--apply") + output = state / "report.json" + else: + request_path = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_ALLOCATION_REQUEST")) + if not _within(request_path, [state]): + raise ValueError("allocation request must be in accepted private runtime state") + request, _ = _json(request_path) + if request.get("schema") != "custodian.weekly-allocation-request.v1" or not request.get("as_of"): + raise ValueError("allocation request schema/as_of unbound") + paths = {} + for name in ("cluster", "allocation", "telemetry", "inventory", "provenance", "previous", "reviews", "output"): + value = request.get(name) + if not value and name in {"previous", "reviews"}: + continue + if not isinstance(value, str) or not value: + raise ValueError("allocation input/output unbound") + candidate = request_path.parent / value + paths[name] = _path(str(candidate), existing=name != "output") + if not _within(paths[name], [state] if name == "output" else [state, fleet, root]): + raise ValueError("allocation input/output outside accepted runtime roots") + output = paths["output"] + if output == request_path: + raise ValueError("allocation output aliases its request") + if output in paths.values() and any(output == path for key, path in paths.items() if key != "output"): + raise ValueError("allocation output aliases an input") + argv += ["--request", str(request_path)] + return argv, root, output, request + + +def _execute(argv: list[str], cwd: Path, timeout: int) -> int: + # Fixed executable and argv, no shell; kill the group so child helper scripts + # cannot outlive a timed-out parent. Never persist child stdout/stderr. + with subprocess.Popen(argv, cwd=cwd, stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + start_new_session=True) as process: + try: + return process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + process.wait() + raise RuntimeError("Custodian operation timed out") from None + + +def run_custodian_operation(operation: str, params: dict) -> dict: + argv, cwd, output, request = build_command(operation, params) + snapshot = None + try: + if request is not None: + # Execute the validated request bytes, not a mutable producer pointer. + with tempfile.NamedTemporaryFile(mode="w", dir=Path(argv[-1]).parent, + suffix=".json", delete=False) as stream: + snapshot = Path(stream.name) + json.dump(request, stream) + argv[-1] = str(snapshot) + started = time.time_ns() + code = _execute(argv, cwd, params.get("timeout_seconds", 240)) + finally: + if snapshot is not None: + snapshot.unlink(missing_ok=True) + if code != 0: + raise RuntimeError(f"Custodian operation failed with exit {code}; inspect owner state") + output = _path(str(output)) + data, raw = _json(output) + if operation == "custodian_security_flow": + if output.stat().st_mtime_ns < started: + raise RuntimeError("Custodian report was not refreshed") + if not all(isinstance(data.get(key), list) for key in ("READY", "UNBLOCKED", "ATTENDED", "STALLED", "DEPENDENT")): + raise RuntimeError("Custodian security report shape invalid") + counts = {key.lower(): len(data[key]) for key in ("READY", "UNBLOCKED", "ATTENDED", "STALLED", "DEPENDENT")} + else: + if data.get("schema") != "custodian.weekly-allocation.v1" or not isinstance(data.get("recommendations"), list): + raise RuntimeError("Custodian allocation report shape invalid") + counts = {"recommendations": len(data["recommendations"])} + return {"kind": operation, "status": "completed", "exit_code": code, + "report_sha256": hashlib.sha256(raw).hexdigest(), "counts": counts, + "apply": params.get("apply", False), + "owner_acknowledgment": "not established by command completion"} diff --git a/tests/test_bounded_operations.py b/tests/test_bounded_operations.py index 31b4b0f..39f51d3 100644 --- a/tests/test_bounded_operations.py +++ b/tests/test_bounded_operations.py @@ -23,6 +23,9 @@ def test_registry_names_only_the_accepted_operations() -> None: "repo_manager_freshness", # ACTIVITY-WP-0043-T06 (operator-triggered single-repo pickup) "owner_return_pickup", + # CUST-WP-0071/0079 fixed owner commands, definitions disabled. + "custodian_security_flow", + "custodian_allocation_review", } diff --git a/tests/test_custodian_operations.py b/tests/test_custodian_operations.py new file mode 100644 index 0000000..06d826c --- /dev/null +++ b/tests/test_custodian_operations.py @@ -0,0 +1,170 @@ +import json +from pathlib import Path +from unittest.mock import patch + +import pytest + +from activity_core import custodian_operations as runner +from activity_core.bounded_operations import normalize_operation_result, validate_bounded_operations +from activity_core.definition_parser import parse_file + + +@pytest.fixture +def runtime(tmp_path, monkeypatch): + root, fleet, state = (tmp_path / x for x in ("owner", "fleet", "state")) + for path in (root / "scripts", fleet, state, state / "security-flow"): + path.mkdir(parents=True, exist_ok=True) + state.chmod(0o700) + (state / "security-flow").chmod(0o700) + for key, value in {"ROOT": root, "FLEET_ROOT": fleet, "STATE_ROOT": state}.items(): + monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_" + key, str(value)) + monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_HUB", "http://127.0.0.1:8000") + monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_REPOS", "the-custodian") + (fleet / "the-custodian/workplans").mkdir(parents=True) + for name in ("security_flow_daily.py", "weekly_allocation_report.py"): + (root / "scripts" / name).write_text("# owner fixture\n") + return root, fleet, state + + +def request(runtime, monkeypatch): + root, fleet, state = runtime + inputs = {} + for key in ("cluster", "allocation", "telemetry", "inventory", "provenance"): + path = fleet / (key + ".json") + path.write_text("{}") + inputs[key] = str(path) + data = {"schema": "custodian.weekly-allocation-request.v1", "as_of": "2026-10-19T00:00:00Z", + **inputs, "output": str(state / "week.json")} + path = state / "request.json" + path.write_text(json.dumps(data)) + monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_ALLOCATION_REQUEST", str(path)) + return path, data + + +def test_fixed_daily_argv_and_missing_binding_before_execution(runtime, monkeypatch): + root, fleet, state = runtime + argv, cwd, output, request_data = runner.build_command("custodian_security_flow", {"apply": True}) + assert argv[1:] == [str(root / "scripts/security_flow_daily.py"), "--root", str(fleet), "--repos", "the-custodian", + "--hub", "http://127.0.0.1:8000", "--state-dir", str(state / "security-flow"), "--apply"] + monkeypatch.delenv("ACTIVITY_CORE_CUSTODIAN_HUB") + with patch.object(runner, "_execute") as execute, pytest.raises(ValueError): + runner.run_custodian_operation("custodian_security_flow", {"apply": True}) + execute.assert_not_called() + + +@pytest.mark.parametrize("params", [{"apply": "true"}, {"apply": True, "command": "echo x"}, + {"apply": True, "timeout_seconds": True}, {"apply": True, "timeout_seconds": 241}, + {"apply": True, "root": "/tmp"}]) +def test_invalid_definition_parameters_refused(params): + with pytest.raises(ValueError): + runner.validate_params("custodian_security_flow", params) + + +def test_definition_no_profile_no_tasks_and_evidence_required(): + root = Path(__file__).resolve().parents[1] + for name in ("daily-security-flow", "weekly-railiance-allocation-review"): + definition = parse_file(root / "activity-definitions" / f"{name}.md") + assert definition.enabled is False + assert definition.rules == definition.instructions == [] + sources = definition.context_sources + validate_bounded_operations(sources, []) + sources[0]["params"].pop("evidence_sinks") + with pytest.raises(ValueError): + validate_bounded_operations(sources, []) + + +def test_symlink_state_and_output_refused(runtime, monkeypatch, tmp_path): + root, fleet, state = runtime + path, data = request(runtime, monkeypatch) + external = tmp_path / "outside" + external.mkdir() + (state / "redirect").symlink_to(external, target_is_directory=True) + data["output"] = str(state / "redirect/week.json") + path.write_text(json.dumps(data)) + with pytest.raises(ValueError, match="symlink"): + runner.build_command("custodian_allocation_review", {}) + data["output"] = str(external / "week.json") + path.write_text(json.dumps(data)) + with pytest.raises(ValueError, match="outside"): + runner.build_command("custodian_allocation_review", {}) + + +def test_unbound_request_and_alias_input_refused(runtime, monkeypatch): + path, data = request(runtime, monkeypatch) + data["cluster"] = None + path.write_text(json.dumps(data)) + with pytest.raises(ValueError, match="unbound"): + runner.build_command("custodian_allocation_review", {}) + path, data = request(runtime, monkeypatch) + data["output"] = str(path) + path.write_text(json.dumps(data)) + with pytest.raises(ValueError, match="request"): + runner.build_command("custodian_allocation_review", {}) + + +def test_fixed_fixture_execution_retains_only_compact_receipt(runtime): + root, fleet, state = runtime + (root / "scripts/security_flow_daily.py").write_text(''' +import argparse,json,pathlib +p=argparse.ArgumentParser() +p.add_argument('--root'); p.add_argument('--repos'); p.add_argument('--hub'); p.add_argument('--state-dir'); p.add_argument('--apply',action='store_true') +a=p.parse_args() +pathlib.Path(a.state_dir,'report.json').write_text(json.dumps({k:[] for k in ['READY','UNBLOCKED','ATTENDED','STALLED','DEPENDENT']})) +print('fixture-output-must-not-enter-receipt') +''') + receipt = runner.run_custodian_operation("custodian_security_flow", {"apply": False}) + assert receipt["status"] == "completed" + assert receipt["counts"]["ready"] == 0 + assert "fixture-output" not in json.dumps(receipt) + assert len(receipt["report_sha256"]) == 64 + + +def test_nonzero_exit_is_not_success_and_error_is_sanitized(runtime): + with patch.object(runner, "_execute", return_value=7), pytest.raises(RuntimeError, match="exit 7"): + runner.run_custodian_operation("custodian_security_flow", {"apply": False}) + + +def test_stale_report_is_not_command_success(runtime): + _, _, state = runtime + output = state / "security-flow/report.json" + output.write_text('{}') + with patch.object(runner, "_execute", return_value=0), pytest.raises(RuntimeError, match="not refreshed"): + runner.run_custodian_operation("custodian_security_flow", {"apply": False}) + + +def test_weekly_fixture_execution_and_receipt_projection(runtime, monkeypatch): + root, _, state = runtime + request(runtime, monkeypatch) + (root / "scripts/weekly_allocation_report.py").write_text(''' +import argparse,json,pathlib +p=argparse.ArgumentParser();p.add_argument('--request');a=p.parse_args() +r=json.loads(pathlib.Path(a.request).read_text()) +pathlib.Path(r['output']).write_text(json.dumps({'schema':'custodian.weekly-allocation.v1','recommendations':[]})) +''') + result = runner.run_custodian_operation("custodian_allocation_review", {}) + assert result["counts"] == {"recommendations": 0} + assert "secret" not in normalize_operation_result("custodian_allocation_review", result | {"secret": "discard"}) + + +@pytest.mark.asyncio +async def test_temporal_dispatch_uses_existing_stage(monkeypatch): + from activity_core.activities import execute_bounded_operation + monkeypatch.setattr(runner, "run_custodian_operation", lambda operation, params: { + "kind": operation, "status": "completed", "counts": {}, "unknown": "discard"}) + result = await execute_bounded_operation({"source": {"type": "shell", + "query": "custodian_allocation_review", "operation": "custodian_allocation_review", + "params": {}, "bind_to": "context.custodian_result"}}) + assert result["custodian_result"]["status"] == "completed" + assert "unknown" not in result["custodian_result"] + + +def test_subprocess_timeout_kills_group(): + import subprocess + process = __import__('unittest.mock', fromlist=['MagicMock']).MagicMock() + process.__enter__.return_value = process + process.pid = 12345 + process.wait.side_effect = [subprocess.TimeoutExpired('fixed-owner-command', 1), 0] + with patch.object(runner.subprocess, 'Popen', return_value=process), patch.object(runner.os, 'killpg') as kill: + with pytest.raises(RuntimeError, match='timed out'): + runner._execute(['fixed-owner-command'], Path('/tmp'), 1) + kill.assert_called_once_with(12345, runner.signal.SIGKILL) diff --git a/workplans/ACTIVITY-WP-0047-daily-security-flow-adoption.md b/workplans/ACTIVITY-WP-0047-daily-security-flow-adoption.md index 216df8c..47040b8 100644 --- a/workplans/ACTIVITY-WP-0047-daily-security-flow-adoption.md +++ b/workplans/ACTIVITY-WP-0047-daily-security-flow-adoption.md @@ -8,7 +8,7 @@ status: active owner: claude-activity-core topic_slug: activity-core created: "2026-10-13" -updated: "2026-10-13" +updated: "2026-10-19" flavor: implementation related: [CUST-WP-0079] state_hub_workstream_id: "ee97869b-dca2-54c9-aa18-0b479c244fb3" @@ -19,10 +19,13 @@ Origin: the-custodian handed over the daily security-flow lane (message user systemd bus is unavailable, so no host timer can run it. Founder accepted adoption on 2026-10-13. -Boundary: Activity Core owns the schedule and the ops_run record. the-custodian -owns `scripts/security_flow_daily.py`, its report and delivery checkpoints. A -rein runs the fixed command; Activity Core does not execute it (SCOPE: no -general execution). The run carries no repository grant. +Boundary, corrected 2026-10-19: Activity Core owns the schedule, bounded +operation activity/context and progress-sink evidence. the-custodian owns +scripts/security_flow_daily.py, report and delivery checkpoints. The fixed +routine uses ACT-ADR-007 existing registered owner-operation exception, not +general coding execution or a rein task. No repository grant is implied. +The original ops_run/profile mechanism remains historical T01 preparation +and is superseded by T04; T02/T03 use the corrected receipt contract. ## Source definition @@ -47,10 +50,13 @@ priority: high state_hub_task_id: "d0d5fddd-2d73-5156-aac3-49837aec2988" ``` -Wait: the claiming rein (rein-aharness, workstation profile) confirms it can -run the command against workstation checkouts with hub access. Then flip -`enabled: true`, release through the normal image/GitOps path, admin-sync, and -run one manual trigger. +Wait: executor owner accepts the bounded-operation runtime in +docs/custodian-bounded-operations.md: pinned owner source/dependencies, +scoped fleet snapshots/repository list, private state outside source roots, +admitted Hub access and checkpointed command behavior. Then enable through +the normal release/admin-sync path and run one manual trigger. The historical +coding-profile route cannot execute this command and must not be activated. +T04 supplies local source integration, not runtime acceptance. ## First unassisted fire @@ -61,6 +67,29 @@ priority: medium state_hub_task_id: "d692a3eb-7878-5405-931c-e11db494f803" ``` -Observe one scheduled fire end to end (ops_run completed, report written, no -duplicate deliveries) and tell the-custodian so CUST-WP-0079-T03 can close. +Observe one scheduled fire end to end: bounded activity completed, refreshed +owner report and compact receipt retained, progress sink outcome verified and +no duplicate owner deliveries. Return evidence to the-custodian for +CUST-WP-0079-T03; exercise missed/failed-run detection and recovery. This updated +operation emits no ops_run. Command success is not delivery or acknowledgment. Wait: T02. + + +## Fixed owner-command integration + +```task +id: ACTIVITY-WP-0047-T04 +status: done +priority: high +``` + +Completed 2026-10-19: the historical harness.agent-dev-local@1.1.1 pin forbids +Python under green-commit-only and denies Hub egress. Replaced its disabled rule +with a registered ACT-ADR-007 operation using shared fixed-command execution, +explicit runtime bindings and existing compact evidence sinks. Weekly +CUST-WP-0071-T05 consumes the wrapper through its own disabled definition. +Fixture tests cover admission, redirection, unbound input, fixed argv, failed +and stale output, sanitized receipts and Temporal dispatch. T01 remains the +historical initial definition outcome; T04 records the distinct compatibility +repair. No schedule, profile widening, real command or credential access was +performed. T02/T03 retain operational acceptance.