#!/usr/bin/env python3 """One documented, validated contract for toggling PRD Plugin (REQ-070). Every PRD Plugin monitor and mode is already a key in .prd_plugin/config.json. This exposes them as a canonical registry (TOGGLES) plus get/set/list so an external app has a single, stable surface instead of poking raw JSON and guessing key paths. `list --json` emits the registry with current values — the machine-consumable contract for other apps. Writes preserve the config file's indentation, line endings, and BOM, and set validates values against the registry (types + allowed enum values). """ from __future__ import annotations import argparse import copy import json import os import re import sys import tempfile from pathlib import Path # The canonical toggle registry. Each entry: the dotted config key, its type, # default, what it controls, and (for enums) the allowed values. REPORTING_TASKS = [ "report_summary", "session_summary", "wiki_synthesis", "triage_draft", ] SUBSTRATE_CAPABILITIES = [ "records", "graph", "events", "discovery", "knowledge", "memory", "context", "skillbooks", "impact", "models", "goals", "bus", "workers", "specialists", "app_server", "diagnostics", "telemetry", "threads", "watches", "federation", "workspace_export", "mutations", "delegated_reporting", "tracking_branches", "verification", "workflow_judgment", ] VERIFICATION_FULL_SUITE_TRIGGERS = [ "release", "core_change", "impact_unavailable", "impact_degraded", "unmapped_change", ] WORKFLOW_IDS = [ "session.start", "session.stop", "request.intake", "request.import", "planning.requirements", "engineering.debug", "engineering.code-review", "engineering.verify", "evidence.record", "project.closeout", "project.maintenance", "install.update-check", "hub.release", ] TOGGLES = [ {"key": "hooks.enabled", "type": "bool", "default": True, "controls": "Master switch for all PRD Plugin hook behavior."}, {"key": "hooks.nudge.enabled", "type": "bool", "default": True, "controls": "Allow the routing nudge on configured host events."}, {"key": "hooks.nudge.on_session_start", "type": "bool", "default": True, "controls": "Inject the routing nudge when a host session starts."}, {"key": "hooks.nudge.on_user_prompt", "type": "bool", "default": True, "controls": "Inject the routing nudge and update checks on every submitted prompt."}, {"key": "hooks.stop_guard.enabled", "type": "bool", "default": True, "controls": "Allow the autonomous run-until-done Stop guard."}, {"key": "hooks.session_report.enabled", "type": "bool", "default": True, "controls": "Generate the configured session report work on Stop."}, {"key": "hooks.session_report.skill_usage_report", "type": "bool", "default": True, "controls": "Regenerate the local skill-usage report during session reporting."}, {"key": "hooks.drift_check.enabled", "type": "bool", "default": True, "controls": "Allow the Stop dispatcher to run the configured drift check."}, {"key": "hooks.reflection.enabled", "type": "bool", "default": True, "controls": "Allow the Stop dispatcher to run configured reflections."}, {"key": "hooks.precommit_gate.enabled", "type": "bool", "default": True, "controls": "Allow the pre-tool dispatcher to evaluate the commit gate."}, {"key": "hooks.test_scope_guard.enabled", "type": "bool", "default": True, "controls": "Block a whole-suite test run mid-work: full tests belong before a " "commit, scoped tests everywhere else."}, {"key": "hooks.skill_log.enabled", "type": "bool", "default": True, "controls": "Allow post-tool skill-usage logging."}, {"key": "hooks.archive_automation.enabled", "type": "bool", "default": True, "controls": "Allow Stop/idle automation-session archiving."}, {"key": "hooks.workflow.enabled", "type": "bool", "default": True, "controls": "Route supported SessionStart and Stop lifecycle orchestration through the deterministic workflow engine."}, {"key": "hooks.workflow.on_session_start", "type": "bool", "default": True, "controls": "Execute the session.start deterministic workflow before SessionStart handlers."}, {"key": "hooks.workflow.on_stop", "type": "bool", "default": True, "controls": "Execute the session.stop deterministic workflow before Stop handlers."}, {"key": "hooks.workflow.fail_closed", "type": "bool", "default": False, "controls": "Block the hook event when its deterministic workflow fails instead of surfacing a fail-open diagnostic."}, {"key": "reasoning_guard.mode", "type": "enum", "allowed": ["on", "report", "off"], "default": "report", "controls": "Observe evidence follow-through across host events: on may block guarded boundaries, report records findings only, and off is inert."}, {"key": "reasoning_guard.backfill.mode", "type": "enum", "allowed": ["live", "recent", "full"], "default": "recent", "controls": "Choose whether Reason Guard starts from the latest visible summary, backfills the configured recent count, or scans the complete trusted rollout history."}, {"key": "reasoning_guard.backfill.limit", "type": "int", "default": 20, "min": 1, "max": 100000, "controls": "Number of historical visible summaries selected when reasoning_guard.backfill.mode is recent."}, {"key": "reasoning_guard.categories.evidence_follow_through", "type": "enum", "allowed": ["inherit", "on", "report", "off"], "default": "inherit", "controls": "Override how unresolved promised checks are handled at a completion boundary."}, {"key": "reasoning_guard.categories.causal_claims", "type": "enum", "allowed": ["inherit", "on", "report", "off"], "default": "inherit", "controls": "Override how causal claims made before promised evidence is resolved are handled."}, {"key": "reasoning_guard.categories.permanent_mutations", "type": "enum", "allowed": ["inherit", "on", "report", "off"], "default": "inherit", "controls": "Override the guard for permanent routing, policy, configuration, and fallback mutations."}, {"key": "reasoning_guard.categories.completion_claims", "type": "enum", "allowed": ["inherit", "on", "report", "off"], "default": "inherit", "controls": "Override how completion claims made while promised evidence remains open are handled."}, {"key": "journal.schema_version", "type": "int", "default": 1, "controls": "Pinned version of the agent journal contract; a reader that does not know this version refuses rather than guesses."}, {"key": "journal.enabled", "type": "bool", "default": False, "controls": "Master switch for the agent journal. Off means no journal operation runs."}, {"key": "journal.capture.automatic", "type": "bool", "default": False, "controls": "Capture entries automatically from filtered host events rather than only on explicit request."}, {"key": "journal.reflections.enabled", "type": "bool", "default": False, "controls": "Allow journal entries to carry typed links to reflection questions and their supporting evidence."}, {"key": "journal.prompt_recall.enabled", "type": "bool", "default": False, "controls": "Allow recall of prior journal entries into agent context."}, {"key": "journal.retention.mode", "type": "enum", "allowed": ["archive-only"], "default": "archive-only", "controls": "How journal segments age out. Archive-only by design: rotation never deletes unacknowledged data and there is no time-based destructive deletion."}, {"key": "journal.security.protected_read_audit", "type": "bool", "default": True, "controls": "Record an audit entry whenever protected journal content is read."}, {"key": "automation.autonomy_level", "type": "enum", "allowed": ["autonomous", "key_decision", "guided"], "default": "key_decision", "controls": "How much the agent decides vs. asks; the ship-consent tier."}, {"key": "automation.autonomous_run_until_done", "type": "bool", "default": True, "controls": "The Stop guard keeps working toward the active goal in autonomous."}, {"key": "automation.key_decision_continue_guard", "type": "bool", "default": True, "controls": "In key_decision, the Stop guard blocks a stop that has not been declared while a goal is open — continuing planned work is not a key decision. Declare a real need in .prd_plugin/local/autonomy-pause to stop."}, {"key": "automation.autonomous_continue_cap", "type": "int", "default": 25, "controls": "Max consecutive auto-continues per session before allowing a stop."}, {"key": "automation.stop_guard_goal_max_age_days", "type": "int", "default": 2, "controls": "Freshness window in days: an open goal older than this only drives " "run-until-done for the session that owns it; stale goals never conscript."}, {"key": "automation.precommit_gate", "type": "bool", "default": False, "controls": "Block git commit when the gate has error-severity findings."}, {"key": "automation.graph_auto_refresh", "type": "bool", "default": False, "controls": "Regenerate the traceability graph on Stop."}, {"key": "automation.version_check.enabled", "type": "bool", "default": True, "controls": "Auto-check npm for a newer plugin version (cached, fail-open)."}, {"key": "automation.version_check.ttl_hours", "type": "int", "default": 1, "controls": "How long the version-check result is cached before re-querying npm."}, {"key": "drift.monitoring.enabled", "type": "bool", "default": True, "controls": "The drift monitor (validators + snapshots/events)."}, {"key": "drift.monitoring.on_stop", "type": "bool", "default": False, "controls": "Run a cheap drift check on every Stop and surface a summary."}, {"key": "drift.monitoring.export.enabled", "type": "bool", "default": False, "controls": "Also append drift events to a committed/shared feed."}, {"key": "drift.monitoring.export.path", "type": "string", "default": ".prd_plugin/drift/events.jsonl", "controls": "Where the exported (committed) drift-event feed is written."}, {"key": "reflection.enabled", "type": "bool", "default": False, "controls": "Enable the configurable reflection-question system."}, {"key": "reflection.on_stop", "type": "bool", "default": True, "controls": "Ask enabled reflection questions when a Stop hook fires."}, {"key": "reflection.max_questions_per_stop", "type": "int", "default": 5, "min": 1, "max": 50, "controls": "Maximum enabled reflection questions included in one Stop pass."}, {"key": "knowledge.llm_wiki.enabled", "type": "bool", "default": True, "controls": "The LLM wiki (query-before-re-derive, ingest-on-close, backfill)."}, {"key": "knowledge.llm_wiki.require_inline_md_links", "type": "bool", "default": True, "controls": "Require resolvable local Markdown references in wiki prose to be inline links."}, {"key": "tracking.branching.enabled", "type": "bool", "default": True, "controls": "Allow lead-created agent tracking branches and serialized canonical promotion."}, {"key": "tracking.branching.require_for_parallel_agents", "type": "bool", "default": True, "controls": "Require parallel workers to update assigned branches instead of canonical tracking state."}, {"key": "workflows.enabled", "type": "bool", "default": True, "controls": "Master switch for deterministic PRD Plugin workflow execution."}, {"key": "workflows.catalog_path", "type": "string", "default": ".prd_plugin/workflows.json", "min_length": 1, "controls": "Repo-relative managed workflow catalog path."}, {"key": "workflows.run_state_path", "type": "string", "default": ".prd_plugin/state/workflow-runs.json", "min_length": 1, "controls": "Repo-relative persistent workflow run and receipt state."}, {"key": "workflows.enabled_ids", "type": "string_list", "allowed_items": WORKFLOW_IDS, "default": WORKFLOW_IDS.copy(), "controls": "Shipped workflows allowed to plan or execute in this repository."}, {"key": "workflows.allow_custom_definitions", "type": "bool", "default": False, "controls": "Allow catalog entries explicitly marked as custom; embedded code remains forbidden."}, {"key": "workflows.allow_state_mutations", "type": "bool", "default": True, "controls": "Allow shipped workflows to invoke allowlisted canonical state mutation tools."}, {"key": "workflows.max_attempts", "type": "int", "default": 5, "min": 1, "max": 20, "controls": "Maximum explicit attempts for safely retryable failed workflow runs."}, {"key": "workflows.max_output_chars", "type": "int", "default": 50000, "min": 1000, "max": 1000000, "controls": "Maximum persisted characters per action output before retaining only its hash and bounded summary."}, {"key": "workflows.judgment.executor", "type": "enum", "allowed": ["ai-collab"], "default": "ai-collab", "controls": "External runtime that executes provider-neutral workflow judgment requests."}, {"key": "workflows.judgment.profile", "type": "string", "default": "fast-capable", "min_length": 1, "controls": "Portable model profile for workflow judgment steps, resolved by the external runtime."}, {"key": "workflows.judgment.require_source_refs", "type": "bool", "default": True, "controls": "Reject judgment requests and results without bounded source provenance."}, {"key": "workflows.judgment.fallback", "type": "enum", "allowed": ["fail", "main"], "default": "fail", "controls": "Behavior when the configured external judgment executor is unavailable."}, {"key": "fork.version_check.enabled", "type": "bool", "default": False, "controls": "Check a configured fork/upstream source for a newer version and surface it on Stop."}, {"key": "fork.version_check.source_url", "type": "string", "default": "", "controls": "JSON URL to poll for the latest fork version (a static manifest or a node /v1/status)."}, {"key": "fork.version_check.version_field", "type": "string", "default": "fork_version", "controls": "Dotted JSON path to the version in the fork source response."}, {"key": "fork.version_check.local_marker", "type": "string", "default": "FORK-VERSION", "controls": "Repo-relative file holding this repo's current fork version."}, {"key": "fork.version_check.ttl_hours", "type": "int", "default": 6, "controls": "How long the fork-version result is cached before re-polling the source."}, {"key": "integrations.substrate.enabled", "type": "bool", "default": False, "controls": "Master switch for the AI-Collab Substrate adapter; false overrides every subordinate setting."}, {"key": "integrations.substrate.mode", "type": "enum", "allowed": ["off", "observe", "coordinate"], "default": "off", "controls": "Adapter authority: off, read-only projection, or explicitly allowlisted coordination."}, {"key": "integrations.substrate.capabilities", "type": "string_list", "allowed_items": SUBSTRATE_CAPABILITIES, "default": ["records", "graph", "events"], "controls": "Substrate adapter capabilities allowed when the master switch and mode permit them."}, {"key": "integrations.substrate.contract_version", "type": "int", "default": 2, "min": 2, "max": 2, "controls": "Version of the PRD Plugin to AI-Collab Substrate adapter contract."}, {"key": "integrations.substrate.require_source_refs", "type": "bool", "default": True, "controls": "Require projected/delegated material claims to retain known PRD source references."}, {"key": "integrations.substrate.endpoint", "type": "string", "default": "http://127.0.0.1:47124", "min_length": 1, "controls": "AI-Collab UTCP gateway base URL; local by default and used only when the adapter is enabled."}, {"key": "integrations.substrate.manual_url", "type": "string", "default": "http://127.0.0.1:47124/utcp.json", "min_length": 1, "controls": "Runtime-owned UTCP manual used for exact tool discovery and call templates."}, {"key": "integrations.substrate.knowledge_browse_url", "type": "string", "default": "http://127.0.0.1:5276", "min_length": 1, "controls": "AI-Collab-owned human Knowledge Hub base URL advertised for optional HTML browsing."}, {"key": "integrations.substrate.auth_env", "type": "string", "default": "AI_COLLAB_API_KEY", "min_length": 1, "controls": "Environment-variable name containing an optional runtime API key; the value is never persisted or reported."}, {"key": "integrations.substrate.timeout_seconds", "type": "int", "default": 5, "min": 1, "max": 120, "controls": "Per-request timeout for runtime discovery and tool calls."}, {"key": "integrations.substrate.max_result_bytes", "type": "int", "default": 1048576, "min": 1024, "max": 16777216, "controls": "Maximum accepted runtime response size before the bridge fails closed."}, {"key": "integrations.substrate.cache_ttl_seconds", "type": "int", "default": 60, "min": 0, "max": 86400, "controls": "Freshness window for ignored local runtime discovery cache entries."}, {"key": "integrations.substrate.fallback", "type": "enum", "allowed": ["fail", "local", "skip"], "default": "local", "controls": "Default explicit behavior when an optional runtime capability is unavailable."}, {"key": "integrations.substrate.automation.discovery_on_session_start", "type": "bool", "default": True, "controls": "Run bounded runtime identity, catalog, health, and service preflight at session start."}, {"key": "integrations.substrate.automation.knowledge_recall", "type": "bool", "default": True, "controls": "Allow relevant workflows to query federated Knowledge Hub sources before local fallback."}, {"key": "integrations.substrate.automation.memory_recall", "type": "bool", "default": True, "controls": "Allow relevant workflows to recall provenance-bounded Substrate memories."}, {"key": "integrations.substrate.automation.context_enrichment", "type": "bool", "default": True, "controls": "Allow relevant judgment workflows to request bounded context packs."}, {"key": "integrations.substrate.automation.notices_on_session_start", "type": "bool", "default": False, "controls": "Read unacknowledged Substrate watch notices at session start without acknowledging them."}, {"key": "integrations.substrate.automation.goals_sync", "type": "bool", "default": False, "controls": "Bind and reconcile PRD tracking goals with Substrate goals while PRD remains canonical."}, {"key": "integrations.substrate.automation.telemetry_on_maintenance", "type": "bool", "default": False, "controls": "Read runtime usage/latency/error telemetry during maintenance; never auto-change profiles."}, {"key": "integrations.substrate.automation.reporting_dispatch", "type": "bool", "default": True, "controls": "Dispatch eligible delegated reporting bundles when reporting delegation is enabled."}, {"key": "integrations.substrate.automation.judgment_dispatch", "type": "bool", "default": True, "controls": "Automatically dispatch hash-bound waiting workflow judgments to the runtime."}, {"key": "integrations.substrate.automation.verification_execution", "type": "bool", "default": True, "controls": "Execute impact-scoped verification plans when test scoping and the runtime are enabled."}, {"key": "verification.test_scope.enabled", "type": "bool", "default": True, "controls": "Allow AI-Collab to execute a conservative impact-scoped verification plan."}, {"key": "verification.test_scope.executor", "type": "enum", "allowed": ["ai-collab"], "default": "ai-collab", "controls": "Runtime that enriches the deterministic plan and executes selected tests."}, {"key": "verification.test_scope.neighbor_limit", "type": "int", "default": 12, "min": 1, "max": 100, "controls": "Maximum AI-Collab impact neighbours requested per exact changed file."}, {"key": "verification.test_scope.max_changed_files", "type": "int", "default": 25, "min": 1, "max": 1000, "controls": "Changed-file count above which verification escalates to the full suite."}, {"key": "verification.test_scope.fallback", "type": "enum", "allowed": ["full"], "default": "full", "controls": "Safe behavior for unavailable, degraded, or unmapped focused selection."}, {"key": "verification.test_scope.full_suite_triggers", "type": "string_list", "allowed_items": VERIFICATION_FULL_SUITE_TRIGGERS, "required_items": VERIFICATION_FULL_SUITE_TRIGGERS, "default": VERIFICATION_FULL_SUITE_TRIGGERS.copy(), "controls": "Conditions that require full verification instead of a focused selection."}, {"key": "verification.test_scope.core_paths", "type": "string_list", "default": ["package.json", "package-lock.json", "pyproject.toml", "Cargo.toml", "go.mod", ".github/workflows/**"], "controls": "Repo-relative glob patterns whose changes require full verification."}, {"key": "verification.test_scope.test_patterns", "type": "string_list", "default": ["tests/**", "test/**", "spec/**", "**/*.test.*", "**/*.spec.*", "test_*.py", "*_test.py", "*_test.go", "**/test_*.py", "**/*_test.py", "**/*_test.go"], "controls": "Repo-relative patterns AI-Collab uses when mapping impacted files to tests."}, {"key": "verification.test_scope.max_selected_tests", "type": "int", "default": 200, "min": 1, "max": 5000, "controls": "Focused-test count above which execution widens to the full suite."}, {"key": "verification.test_scope.execution_timeout_seconds", "type": "int", "default": 600, "min": 1, "max": 86400, "controls": "Timeout applied independently to each configured verification command."}, {"key": "verification.test_scope.focused_commands", "type": "json_list", "default": [], "controls": "Optional argv arrays for focused verification; a {tests} item expands to selected test paths."}, {"key": "verification.test_scope.full_commands", "type": "json_list", "default": [], "controls": "Optional argv arrays for full verification; empty uses deterministic language detection."}, {"key": "verification.test_scope.allowed_executables", "type": "string_list", "default": ["python", "python3", "py", "pytest", "npm", "npx", "node", "go", "cargo", "dotnet", "mvn", "gradle", "gradlew", "make"], "controls": "Executable allowlist for configured verification commands; shell expressions are never evaluated."}, {"key": "reporting.delegation.enabled", "type": "bool", "default": False, "controls": "Allow selected non-deterministic reporting tasks to use an external executor."}, {"key": "reporting.delegation.executor", "type": "enum", "allowed": ["ai-collab"], "default": "ai-collab", "controls": "Runtime that resolves the configured profile and executes delegated reporting."}, {"key": "reporting.delegation.contract_version", "type": "int", "default": 1, "min": 1, "max": 1, "controls": "Version of the deterministic reporting bundle/result contract."}, {"key": "reporting.delegation.profile", "type": "string", "default": "fast-capable", "min_length": 1, "controls": "Portable model profile alias resolved by the external executor."}, {"key": "reporting.delegation.tasks", "type": "string_list", "allowed_items": REPORTING_TASKS, "default": REPORTING_TASKS.copy(), "controls": "Non-deterministic reporting tasks eligible for delegation."}, {"key": "reporting.delegation.fallback", "type": "enum", "allowed": ["main", "deterministic_only", "fail"], "default": "main", "controls": "Behavior when delegation is disabled, unavailable, disallowed, or times out."}, {"key": "reporting.delegation.require_source_refs", "type": "bool", "default": True, "controls": "Reject delegated notable items/actions that lack valid deterministic source references."}, {"key": "reporting.delegation.max_input_tokens", "type": "int", "default": 12000, "min": 1000, "max": 200000, "controls": "Maximum input-token budget requested from the delegated executor."}, {"key": "reporting.delegation.max_output_tokens", "type": "int", "default": 1500, "min": 100, "max": 20000, "controls": "Maximum output-token budget requested from the delegated executor."}, {"key": "reporting.delegation.timeout_seconds", "type": "int", "default": 30, "min": 1, "max": 600, "controls": "Executor timeout before applying the configured fallback."}, ] TOGGLES += [ {"key": "fabric.default_when_unmapped", "type": "enum", "allowed": ["raw", "abstain"], "default": "raw", "controls": "Fail-safe for models with no fabric binding: run raw (no prediction " "treatment) or abstain entirely. Never guesses a profile (ai-collab EV-477)."}, {"key": "fabric.binding_requires_evidence", "type": "bool", "default": True, "controls": "Reject fabric model-profile bindings that lack a non-empty evidence " "reference, keeping the map measured rather than hand-typed."}, ] _BY_KEY = {t["key"]: t for t in TOGGLES} _MANAGED_OBJECT_PATHS = { "configuration.custom_profiles": "profiles", "fabric.model_profiles": "set-fabric-binding", # Operator-populated address book: {peer repo id -> path}. Its keys are # repo names, so they can never appear in shipped defaults. "requests.peers": "request_routing", "reflection.categories": "prd_reflections", } _READ_ONLY_KEYS = { "schema_version", "plugin.installed_version", "configuration.active_profile", "configuration.custom_profiles", "fabric.model_profiles", "ids.required_prefixes", "reflection.categories", } _READ_ONLY_PREFIXES = ("automation.autonomy_levels.",) _LATENCY = { "hooks.enabled": "high", "hooks.nudge.enabled": "high", "hooks.nudge.on_session_start": "low", "hooks.nudge.on_user_prompt": "high", "hooks.stop_guard.enabled": "low", "hooks.session_report.enabled": "medium", "hooks.session_report.skill_usage_report": "medium", "hooks.drift_check.enabled": "high", "hooks.reflection.enabled": "medium", "hooks.precommit_gate.enabled": "medium", "hooks.test_scope_guard.enabled": "medium", "hooks.skill_log.enabled": "high", "hooks.archive_automation.enabled": "medium", "hooks.workflow.enabled": "medium", "hooks.workflow.on_session_start": "low", "hooks.workflow.on_stop": "medium", "reasoning_guard.mode": "high", "reasoning_guard.backfill.mode": "high", "reasoning_guard.backfill.limit": "high", "reasoning_guard.categories.evidence_follow_through": "high", "reasoning_guard.categories.causal_claims": "high", "reasoning_guard.categories.permanent_mutations": "high", "reasoning_guard.categories.completion_claims": "high", "workflows.enabled": "medium", "automation.graph_auto_refresh": "high", "automation.version_check.enabled": "medium", "drift.monitoring.enabled": "high", "drift.monitoring.on_stop": "high", "reflection.enabled": "medium", "reflection.on_stop": "medium", "knowledge.llm_wiki.enabled": "low", "reporting.delegation.enabled": "medium", "verification.test_scope.enabled": "high", "integrations.substrate.enabled": "medium", "fork.version_check.enabled": "medium", } _DEPENDENCIES = { "hooks.nudge.enabled": ["hooks.enabled"], "hooks.nudge.on_session_start": ["hooks.enabled", "hooks.nudge.enabled"], "hooks.nudge.on_user_prompt": ["hooks.enabled", "hooks.nudge.enabled"], "hooks.stop_guard.enabled": ["hooks.enabled"], "hooks.session_report.enabled": ["hooks.enabled"], "hooks.session_report.skill_usage_report": ["hooks.enabled", "hooks.session_report.enabled"], "hooks.drift_check.enabled": ["hooks.enabled"], "hooks.reflection.enabled": ["hooks.enabled"], "hooks.precommit_gate.enabled": ["hooks.enabled"], "hooks.test_scope_guard.enabled": ["hooks.enabled"], "hooks.skill_log.enabled": ["hooks.enabled"], "hooks.archive_automation.enabled": ["hooks.enabled"], "hooks.workflow.enabled": ["hooks.enabled", "workflows.enabled"], "hooks.workflow.on_session_start": ["hooks.enabled", "hooks.workflow.enabled", "workflows.enabled"], "hooks.workflow.on_stop": ["hooks.enabled", "hooks.workflow.enabled", "workflows.enabled"], "reasoning_guard.mode": ["hooks.enabled"], "reasoning_guard.backfill.mode": ["hooks.enabled", "reasoning_guard.mode"], "reasoning_guard.backfill.limit": [ "hooks.enabled", "reasoning_guard.mode", "reasoning_guard.backfill.mode", ], "reasoning_guard.categories.evidence_follow_through": ["hooks.enabled"], "reasoning_guard.categories.causal_claims": ["hooks.enabled"], "reasoning_guard.categories.permanent_mutations": ["hooks.enabled"], "reasoning_guard.categories.completion_claims": ["hooks.enabled"], "automation.autonomous_run_until_done": ["hooks.enabled", "hooks.stop_guard.enabled"], "automation.key_decision_continue_guard": ["hooks.enabled", "hooks.stop_guard.enabled"], "automation.precommit_gate": ["hooks.enabled", "hooks.precommit_gate.enabled"], "automation.graph_auto_refresh": ["hooks.enabled", "hooks.session_report.enabled"], "drift.monitoring.on_stop": ["drift.monitoring.enabled", "hooks.enabled", "hooks.drift_check.enabled"], "reflection.on_stop": ["reflection.enabled", "hooks.enabled", "hooks.reflection.enabled"], "journal.capture.automatic": ["journal.enabled"], "journal.reflections.enabled": ["journal.enabled"], "journal.prompt_recall.enabled": ["journal.enabled"], "journal.retention.mode": ["journal.enabled"], "journal.security.protected_read_audit": ["journal.enabled"], } # The journal is a CONTRACT, not seven independent switches: a half-configured # family silently changes what gets captured and what is allowed to be read. # All seven present or none. JOURNAL_CONTRACT_KEYS = ( "journal.schema_version", "journal.enabled", "journal.capture.automatic", "journal.reflections.enabled", "journal.prompt_recall.enabled", "journal.retention.mode", "journal.security.protected_read_audit", ) JOURNAL_SUPPORTED_SCHEMA_VERSIONS = (1,) BUILTIN_PROFILES = { "off": { "description": "Switch PRD Plugin behavior off entirely: no hooks, no stop guard, no gates, no workflows, no telemetry. Skills/tools stay installed but nothing runs automatically. Reverse with any other profile (/prd-on).", "settings": { "hooks.enabled": False, "hooks.nudge.enabled": False, "hooks.nudge.on_session_start": False, "hooks.nudge.on_user_prompt": False, "hooks.stop_guard.enabled": False, "hooks.session_report.enabled": False, "hooks.session_report.skill_usage_report": False, "hooks.drift_check.enabled": False, "hooks.reflection.enabled": False, "hooks.precommit_gate.enabled": False, "hooks.skill_log.enabled": False, "hooks.archive_automation.enabled": False, "hooks.workflow.enabled": False, "reasoning_guard.mode": "off", "workflows.enabled": False, "automation.precommit_gate": False, "automation.autonomous_run_until_done": False, "automation.graph_auto_refresh": False, "automation.version_check.enabled": False, "drift.monitoring.enabled": False, "drift.monitoring.on_stop": False, "reflection.enabled": False, "knowledge.llm_wiki.enabled": False, "integrations.substrate.enabled": False, "verification.test_scope.enabled": True, "reporting.delegation.enabled": False, "fork.version_check.enabled": False, }, }, "lean": { "description": "Latency-first local work: retain routing at session start and safety gates, disable repeated prompt/Stop telemetry and optional integrations.", "settings": { "hooks.enabled": True, "hooks.nudge.enabled": True, "hooks.nudge.on_session_start": True, "hooks.nudge.on_user_prompt": False, "hooks.stop_guard.enabled": True, "hooks.session_report.enabled": False, "hooks.session_report.skill_usage_report": False, "hooks.drift_check.enabled": False, "hooks.reflection.enabled": False, "hooks.precommit_gate.enabled": True, "hooks.skill_log.enabled": False, "hooks.archive_automation.enabled": False, "hooks.workflow.enabled": False, "reasoning_guard.mode": "off", "workflows.enabled": True, "automation.precommit_gate": True, "automation.graph_auto_refresh": False, "automation.version_check.enabled": False, "drift.monitoring.on_stop": False, "reflection.enabled": False, "knowledge.llm_wiki.enabled": False, "integrations.substrate.enabled": False, "verification.test_scope.enabled": True, "reporting.delegation.enabled": False, "fork.version_check.enabled": False, }, }, "balanced": { "description": "Recommended everyday profile: one session-start nudge, safety gates and bounded Stop reporting, with expensive optional work off.", "settings": { "hooks.enabled": True, "hooks.nudge.enabled": True, "hooks.nudge.on_session_start": True, "hooks.nudge.on_user_prompt": False, "hooks.stop_guard.enabled": True, "hooks.session_report.enabled": True, "hooks.session_report.skill_usage_report": True, "hooks.drift_check.enabled": True, "hooks.reflection.enabled": True, "hooks.precommit_gate.enabled": True, "hooks.skill_log.enabled": False, "hooks.archive_automation.enabled": True, "hooks.workflow.enabled": True, "reasoning_guard.mode": "report", "hooks.workflow.on_session_start": True, "hooks.workflow.on_stop": True, "workflows.enabled": True, "automation.precommit_gate": True, "automation.graph_auto_refresh": False, "automation.version_check.enabled": True, "drift.monitoring.enabled": False, "drift.monitoring.on_stop": False, "reflection.enabled": False, "knowledge.llm_wiki.enabled": True, "integrations.substrate.enabled": False, "verification.test_scope.enabled": True, "reporting.delegation.enabled": False, "fork.version_check.enabled": False, }, }, "thorough": { "description": "Maximum local method feedback: all local hooks, prompt nudges, drift, reflections, reports, graph refresh, and wiki behavior enabled; external integrations remain explicit.", "settings": { "hooks.enabled": True, "hooks.nudge.enabled": True, "hooks.nudge.on_session_start": True, "hooks.nudge.on_user_prompt": True, "hooks.stop_guard.enabled": True, "hooks.session_report.enabled": True, "hooks.session_report.skill_usage_report": True, "hooks.drift_check.enabled": True, "hooks.reflection.enabled": True, "hooks.precommit_gate.enabled": True, "hooks.skill_log.enabled": True, "hooks.archive_automation.enabled": True, "hooks.workflow.enabled": True, "reasoning_guard.mode": "on", "hooks.workflow.on_session_start": True, "hooks.workflow.on_stop": True, "workflows.enabled": True, "automation.precommit_gate": True, "automation.graph_auto_refresh": True, "automation.version_check.enabled": True, "drift.monitoring.enabled": True, "drift.monitoring.on_stop": True, "reflection.enabled": True, "reflection.on_stop": True, "knowledge.llm_wiki.enabled": True, "integrations.substrate.enabled": False, "verification.test_scope.enabled": True, "reporting.delegation.enabled": False, "fork.version_check.enabled": False, }, }, } CONTROL_PLANES = [ {"kind": "environment", "scope": "process/session", "controls": [ "PRD_PLUGIN_ROOT", "PRD_PLUGIN_PYTHON", "PRD_UPSTREAM_HUB", "PRD_WORKER_SESSION", "PRD_TRACKING_BRANCH_ID", "PRD_TRACKING_BRANCH_OWNER", "PRD_STOP_GUARD", "PRD_REFLECTIONS", "CODEX_HOME", ]}, {"kind": "specialized_crud", "scope": "persistent structured config", "controls": [ "reflection.categories", "reflection.questions", ], "manager": "prd_reflections CLI / PRD MCP reflection tools"}, {"kind": "install_time", "scope": "installer invocation", "controls": [ "target host: codex/opencode/claude/both", "force/reset state", "Claude plugin config", "repo-local skill injection", "optional script scopes", ], "manager": "prd-install / prd_install.py"}, {"kind": "host_wiring", "scope": "host lifecycle", "controls": [ ".codex/hooks.json", ".claude/settings.json", ".opencode/plugins/prd-hooks.js", ], "manager": "installer plus prd_hook_dispatch.py"}, {"kind": "invocation_only", "scope": "one command", "controls": [ "repo root", "output/format", "dry-run", "base ref", "limit", "force refresh", "explicit request/message/session identifiers", ], "manager": "individual command argparse interfaces"}, ] _TRUE = {"true", "1", "on", "yes", "enable", "enabled"} _FALSE = {"false", "0", "off", "no", "disable", "disabled"} def _config_path(root): return Path(root) / ".prd_plugin" / "config.json" def _read_raw(path): try: # Decode bytes directly: Path.read_text() performs universal-newline # translation, which would hide CRLF style from _style(). Plain UTF-8 # also leaves a leading BOM visible so the next write can preserve it. return path.read_bytes().decode("utf-8") except OSError: return "" def _style(text): import re bom = text.startswith(chr(0xFEFF)) body = text[1:] if bom else text newline = "\r\n" if "\r\n" in body else "\n" m = re.search(r"[\r\n]([ \t]+)\S", body) if m: found = m.group(1) indent = "\t" if found.startswith("\t") else len(found) else: indent = 2 return indent, newline, bom def _load(root): text = _read_raw(_config_path(root)) if not text.strip(): return {}, (2, "\n", False) style = _style(text) body = text[1:] if style[2] else text try: return json.loads(body), style except json.JSONDecodeError as exc: raise ValueError(f"{_config_path(root)} is not valid JSON: {exc}") from exc def _save(root, data, style): indent, newline, bom = style body = json.dumps(data, indent=indent, ensure_ascii=False) + "\n" if newline != "\n": body = body.replace("\n", newline) encoded = ((chr(0xFEFF) if bom else "") + body).encode("utf-8") path = _config_path(root) path.parent.mkdir(parents=True, exist_ok=True) temp_path = None try: with tempfile.NamedTemporaryFile( mode="wb", dir=path.parent, prefix=path.name + ".", suffix=".tmp", delete=False) as handle: temp_path = Path(handle.name) handle.write(encoded) handle.flush() os.fsync(handle.fileno()) os.replace(temp_path, path) except Exception: if temp_path is not None: try: temp_path.unlink(missing_ok=True) except OSError: pass raise def _dig(data, key): cur = data for part in key.split("."): if not isinstance(cur, dict) or part not in cur: return None, False cur = cur[part] return cur, True def _put(data, key, value): cur = data parts = key.split(".") for part in parts[:-1]: if part not in cur: nxt = {} cur[part] = nxt else: nxt = cur[part] if not isinstance(nxt, dict): raise ValueError(f"cannot set {key}: parent {part!r} is not an object") cur = nxt cur[parts[-1]] = value def _template_candidates(root): root = Path(root) script_root = Path(__file__).resolve().parents[1] return ( root / ".prd_plugin" / "templates" / "config.json", root / "templates" / "config.json", script_root / "templates" / "config.json", script_root / ".prd_plugin" / "templates" / "config.json", ) def _defaults(root): for path in _template_candidates(root): if path.is_file(): try: data = json.loads(path.read_text(encoding="utf-8-sig")) if isinstance(data, dict): return data, path except (OSError, json.JSONDecodeError): continue data, _ = _load(root) return data, _config_path(root) def _flatten(data, prefix=""): if prefix in _MANAGED_OBJECT_PATHS: yield prefix, data return if isinstance(data, dict): if not data and prefix: yield prefix, data return for key, value in data.items(): path = f"{prefix}.{key}" if prefix else key yield from _flatten(value, path) return yield prefix, data def _inferred_type(value): if isinstance(value, bool): return "bool" if isinstance(value, int): return "int" if isinstance(value, float): return "number" if value is None: return "nullable_string" if isinstance(value, list): return "string_list" if all(isinstance(item, str) for item in value) else "json_list" if isinstance(value, dict): return "json_object" return "string" def journal_contract_state(root="."): """Report whether the journal contract is absent, broken, or honoured. Three states, deliberately distinct (ai-collab-v3 REQ-141): - `unavailable` — the family is absent entirely. Not an error: the repo simply has no journal, and health reports it as unavailable. - `invalid` — SOME keys are present. This is a violation, not a partial feature, because a half-configured journal quietly changes what is captured and who may read it. Every non-observational journal operation is refused until the family is complete. - `valid` — all seven present and the schema version is one this build understands. Fails closed: anything short of a complete, understood contract leaves `operations_permitted` False. """ data, _ = _load(root) flat = dict(_flatten(data)) present = [key for key in JOURNAL_CONTRACT_KEYS if key in flat] missing = [key for key in JOURNAL_CONTRACT_KEYS if key not in flat] if not present: return {"state": "unavailable", "operations_permitted": False, "present": [], "missing": list(JOURNAL_CONTRACT_KEYS), "reason": "no journal.* settings are configured; the journal is unavailable"} if missing: return {"state": "invalid", "operations_permitted": False, "present": present, "missing": missing, "reason": ("the journal contract is incomplete, so journal operations are " f"refused; missing: {', '.join(missing)}")} version = flat.get("journal.schema_version") if version not in JOURNAL_SUPPORTED_SCHEMA_VERSIONS: return {"state": "invalid", "operations_permitted": False, "present": present, "missing": [], "reason": (f"journal.schema_version {version!r} is not supported by this build " f"(supported: {', '.join(str(v) for v in JOURNAL_SUPPORTED_SCHEMA_VERSIONS)})")} return {"state": "valid", "operations_permitted": True, "present": present, "missing": [], "reason": "the journal contract is complete and its schema version is supported"} def _activation_for(key): if key.startswith(("hooks.", "automation.", "drift.", "reflection.", "reporting.", "workflows.", "verification.", "integrations.", "fork.", "knowledge.", "journal.")): return "runtime" if key.startswith(("skills.", "script_installation.")): return "install" if key.startswith(("release.", "health.", "requests.", "privacy.")): return "validation" if key.startswith(("schema_version", "plugin.", "ids.")): return "system" return "policy" def _manager_for(key): if key in _MANAGED_OBJECT_PATHS: return _MANAGED_OBJECT_PATHS[key] if key == "plugin.installed_version": return "installer" if key == "ids.required_prefixes": return "id_registry" if key == "schema_version": return "schema" if key == "configuration.active_profile": return "profiles" return "prd_config" def build_catalog(root="."): """Return a complete catalog derived from shipped defaults plus metadata.""" defaults, source = _defaults(root) catalog = {} for key, default in _flatten(defaults): explicit = dict(_BY_KEY.get(key, {})) mutable = key not in _READ_ONLY_KEYS and not key.startswith(_READ_ONLY_PREFIXES) spec = { "key": key, "type": explicit.pop("type", _inferred_type(default)), "default": copy.deepcopy(default), "controls": explicit.pop("controls", f"Persistent PRD Plugin setting: {key}."), "mutable": mutable, "manager": _manager_for(key), "category": key.split(".", 1)[0], "latency": _LATENCY.get(key, "none"), "activation": _activation_for(key), "dependencies": list(_DEPENDENCIES.get(key, [])), "source": str(source).replace("\\", "/"), } spec.update(explicit) if not mutable: spec["mutable"] = False catalog[key] = spec return catalog def _coerce(spec, value): t = spec["type"] if t == "bool": if isinstance(value, bool): return value s = str(value).strip().lower() if s in _TRUE: return True if s in _FALSE: return False raise ValueError(f"{spec['key']} expects a boolean (on/off), got {value!r}") if t == "int": try: result = int(value) except (TypeError, ValueError): raise ValueError(f"{spec['key']} expects an integer, got {value!r}") if "min" in spec and result < spec["min"]: raise ValueError(f"{spec['key']} must be >= {spec['min']}, got {result}") if "max" in spec and result > spec["max"]: raise ValueError(f"{spec['key']} must be <= {spec['max']}, got {result}") return result if t == "enum": s = str(value) if s not in spec["allowed"]: raise ValueError(f"{spec['key']} must be one of {spec['allowed']}, got {value!r}") return s if t == "string_list": if isinstance(value, str): items = [item.strip() for item in value.split(",") if item.strip()] elif isinstance(value, (list, tuple)): items = [str(item).strip() for item in value if str(item).strip()] else: raise ValueError(f"{spec['key']} expects a comma-separated string or list") if len(items) != len(set(items)): raise ValueError(f"{spec['key']} contains duplicate values: {items}") allowed = spec.get("allowed_items") unknown = [item for item in items if allowed is not None and item not in allowed] if unknown: raise ValueError(f"{spec['key']} contains unknown values {unknown}; allowed: {allowed}") required = spec.get("required_items", []) missing = [item for item in required if item not in items] if missing: raise ValueError(f"{spec['key']} is missing required values {missing}") return items if t == "number": if isinstance(value, bool): raise ValueError(f"{spec['key']} expects a number, got {value!r}") try: return float(value) except (TypeError, ValueError): raise ValueError(f"{spec['key']} expects a number, got {value!r}") if t == "nullable_string": if value is None or (isinstance(value, str) and value.strip().lower() in {"null", "none"}): return None return str(value) if t == "json_list": if not isinstance(value, list): raise ValueError(f"{spec['key']} expects a JSON list") return copy.deepcopy(value) if t == "json_object": if not isinstance(value, dict): raise ValueError(f"{spec['key']} expects a JSON object") return copy.deepcopy(value) result = str(value) if len(result) < spec.get("min_length", 0): raise ValueError(f"{spec['key']} must not be empty") return result def get(root, key): catalog = build_catalog(root) if key not in catalog: raise KeyError(f"unknown toggle {key!r}") data, _ = _load(root) value, present = _dig(data, key) return value if present else copy.deepcopy(catalog[key]["default"]) def set_toggle(root, key, value): spec = build_catalog(root).get(key) if spec is None: raise KeyError(f"unknown toggle {key!r}") if not spec.get("mutable"): raise ValueError(f"{key} is managed by {spec.get('manager')} and is read-only here") coerced = _coerce(spec, value) data, style = _load(root) _put(data, key, coerced) if not key.startswith("configuration."): _put(data, "configuration.active_profile", "custom") _save(root, data, style) if key.startswith("integrations.substrate."): try: import prd_services manifest = Path(root).resolve() / ".prd_plugin" / "services.json" if manifest.is_file(): enabled = bool(get(root, "integrations.substrate.enabled")) and get(root, "integrations.substrate.mode") != "off" prd_services.sync_substrate_consumer( root, enabled=enabled, capabilities=list(get(root, "integrations.substrate.capabilities")), contract_version=int(get(root, "integrations.substrate.contract_version")), fallback=str(get(root, "integrations.substrate.fallback")), ) except (ImportError, OSError, ValueError): # Config remains the authority; services.audit reports any manifest # drift instead of making a valid config change non-atomic. pass return coerced def list_toggles(root): data, _ = _load(root) out = [] for spec in TOGGLES: value, present = _dig(data, spec["key"]) entry = dict(spec) entry["value"] = value if present else spec["default"] entry["is_default"] = not present out.append(entry) return out def list_settings(root, *, category=None, latency=None, mutable=None): data, _ = _load(root) rows = [] for key, spec in build_catalog(root).items(): if category and spec["category"] != category: continue if latency and spec["latency"] != latency: continue if mutable is not None and spec["mutable"] is not mutable: continue value, present = _dig(data, key) row = dict(spec) row["value"] = value if present else copy.deepcopy(spec["default"]) row["is_default"] = row["value"] == spec["default"] rows.append(row) return rows def describe(root, key): spec = build_catalog(root).get(key) if spec is None: raise KeyError(f"unknown setting {key!r}") result = dict(spec) result["value"] = get(root, key) return result def enable(root, key): spec = build_catalog(root).get(key) if spec is None: raise KeyError(f"unknown setting {key!r}") if spec["type"] != "bool" or not spec["mutable"]: raise ValueError(f"{key} is not a mutable boolean setting") return set_toggle(root, key, True) def disable(root, key): spec = build_catalog(root).get(key) if spec is None: raise KeyError(f"unknown setting {key!r}") if spec["type"] != "bool" or not spec["mutable"]: raise ValueError(f"{key} is not a mutable boolean setting") return set_toggle(root, key, False) def effective(root, key): spec = describe(root, key) value = spec["value"] blocked = [dependency for dependency in spec["dependencies"] if get(root, dependency) is not True] active = bool(value) and not blocked if spec["type"] == "bool" else not blocked return { "key": key, "configured": value, "effective": active, "blocked_by": blocked, "dependencies": spec["dependencies"], } def _validate_direct_json_type(spec, value): """Reject values that only the convenience CLI coercions would accept.""" kind = spec["type"] valid = True if kind == "bool": valid = isinstance(value, bool) elif kind == "int": valid = isinstance(value, int) and not isinstance(value, bool) elif kind == "number": valid = isinstance(value, (int, float)) and not isinstance(value, bool) elif kind in {"string", "enum"}: valid = isinstance(value, str) elif kind == "nullable_string": valid = value is None or isinstance(value, str) elif kind == "string_list": valid = isinstance(value, list) and all(isinstance(item, str) for item in value) elif kind == "json_list": valid = isinstance(value, list) elif kind == "json_object": valid = isinstance(value, dict) if not valid: raise ValueError(f"{spec['key']} expects JSON type {kind}, got {type(value).__name__}") def audit_config(root="."): catalog = build_catalog(root) defaults, _ = _defaults(root) data, _ = _load(root) default_keys = {key for key, _ in _flatten(defaults)} catalog_keys = set(catalog) current = dict(_flatten(data)) # extensions.* is the repo-owned namespace (REQ-123): custom settings live # there instead of inside plugin namespaces, so they never audit as # unknown. Everything outside it stays strict. unknown = sorted(key for key in current if key not in catalog_keys and key != "extensions" and not key.startswith("extensions.")) missing = sorted(key for key in catalog_keys if not _dig(data, key)[1]) invalid = [] for key, spec in catalog.items(): value, present = _dig(data, key) if not present: continue try: _validate_direct_json_type(spec, value) _coerce(spec, value) except (TypeError, ValueError) as exc: invalid.append({"key": key, "error": str(exc), "value": value}) profile_names = set(BUILTIN_PROFILES) try: profiles = _profile_map(root) profile_names.update(profiles) for name, profile in profiles.items(): if not profile.get("builtin"): _validated_profile_settings(root, name) except (KeyError, TypeError, ValueError) as exc: value, _ = _dig(data, "configuration.custom_profiles") invalid.append({"key": "configuration.custom_profiles", "error": str(exc), "value": value}) invalid.extend(_fabric_binding_errors(data)) if "extensions" in data and not isinstance(data.get("extensions"), dict): invalid.append({"key": "extensions", "error": "extensions must be an object of repo-owned settings", "value": data.get("extensions")}) active_profile, active_present = _dig(data, "configuration.active_profile") if (active_present and isinstance(active_profile, str) and active_profile != "custom" and active_profile not in profile_names): invalid.append({"key": "configuration.active_profile", "error": f"unknown active profile {active_profile!r}", "value": active_profile}) unclassified = sorted(default_keys - catalog_keys) summary = { "settings": len(catalog), "mutable": sum(1 for spec in catalog.values() if spec["mutable"]), "read_only_or_managed": sum(1 for spec in catalog.values() if not spec["mutable"]), "unclassified": len(unclassified), "unknown": len(unknown), "invalid": len(invalid), "missing_using_defaults": len(missing), } return { "status": "ok" if not (unclassified or unknown or invalid) else "error", "summary": summary, "unclassified_keys": unclassified, "unknown_keys": unknown, "invalid_values": invalid, "missing_using_defaults": missing, } def inventory(root="."): audit = audit_config(root) rows = list_settings(root) by_latency = {level: sum(1 for row in rows if row["latency"] == level) for level in ("high", "medium", "low", "none")} return { "summary": { "persistent_settings": len(rows), "mutable_settings": sum(1 for row in rows if row["mutable"]), "managed_or_read_only_settings": sum(1 for row in rows if not row["mutable"]), "unclassified_settings": audit["summary"]["unclassified"], "latency": by_latency, }, "settings": rows, "control_planes": copy.deepcopy(CONTROL_PLANES), "audit": audit, } def _fabric_binding_errors(data): """Validation findings for fabric.model_profiles (REQ-120). The map shape is {model: {task_type|'*': {profile, evidence}}}; while fabric.binding_requires_evidence is true a binding without a non-empty evidence reference is invalid - bindings stay measured, never hand-typed.""" errors = [] fabric = data.get("fabric") or {} if not isinstance(fabric, dict): return [{"key": "fabric", "error": "fabric must be an object", "value": fabric}] bindings = fabric.get("model_profiles", {}) require_evidence = fabric.get("binding_requires_evidence", True) if not isinstance(bindings, dict): return [{"key": "fabric.model_profiles", "error": "fabric.model_profiles must be an object mapping model ids", "value": bindings}] for model, per_task in bindings.items(): prefix = f"fabric.model_profiles.{model}" if not isinstance(per_task, dict): errors.append({"key": prefix, "error": "model entry must map task types to bindings", "value": per_task}) continue for task_type, binding in per_task.items(): key = f"{prefix}.{task_type}" if not isinstance(binding, dict): errors.append({"key": key, "error": "binding must be an object", "value": binding}) continue profile = binding.get("profile") if not isinstance(profile, str) or not profile.strip(): errors.append({"key": key, "error": "binding requires a non-empty profile id", "value": binding}) evidence = binding.get("evidence") if require_evidence and (not isinstance(evidence, str) or not evidence.strip()): errors.append({"key": key, "error": "binding requires a non-empty evidence reference " "while fabric.binding_requires_evidence is true", "value": binding}) return errors def resolve_fabric_profile(root, model, task_type=None): """Resolve the fabric prediction treatment for a model (REQ-120). Precedence: exact task_type binding, else the model-wide '*' binding, else the fabric.default_when_unmapped fail-safe - {'action': 'raw'} means run with NO prediction treatment; a profile is never guessed (ai-collab EV-477: the wrong treatment regresses below raw).""" data, _ = _load(root) fabric = data.get("fabric") or {} bindings = fabric.get("model_profiles") if isinstance(fabric, dict) else {} per_task = bindings.get(model) if isinstance(bindings, dict) else None if isinstance(per_task, dict): candidates = (task_type, "*") if task_type else ("*",) for candidate in candidates: binding = per_task.get(candidate) if isinstance(binding, dict) and str(binding.get("profile", "")).strip(): return {"action": "apply", "profile": binding["profile"], "evidence": binding.get("evidence"), "source": "binding"} default = fabric.get("default_when_unmapped", "raw") if isinstance(fabric, dict) else "raw" return {"action": default if default in ("raw", "abstain") else "raw", "source": "default"} def set_fabric_binding(root, model, profile, evidence=None, task_type=None): """Mint one evidence-bound fabric binding (fail-closed merge, REQ-120). Intended primary caller: a calibration harness promoting a measured profile, so validation refuses unmeasured or unevidenced writes.""" if not isinstance(model, str) or not model.strip(): raise ValueError("model is required") if not isinstance(profile, str) or not profile.strip(): raise ValueError("profile is required") data, style = _load(root) require_evidence = bool(get(root, "fabric.binding_requires_evidence")) if require_evidence and (not isinstance(evidence, str) or not evidence.strip()): raise ValueError("an evidence reference is required while " "fabric.binding_requires_evidence is true") fabric = data.setdefault("fabric", {}) if not isinstance(fabric, dict): raise ValueError("fabric must be an object") bindings = fabric.setdefault("model_profiles", {}) if not isinstance(bindings, dict): raise ValueError("fabric.model_profiles must be an object") entry = {"profile": profile.strip()} if isinstance(evidence, str) and evidence.strip(): entry["evidence"] = evidence.strip() task_key = task_type.strip() if isinstance(task_type, str) and task_type.strip() else "*" bindings.setdefault(model.strip(), {})[task_key] = entry errors = _fabric_binding_errors(data) if errors: raise ValueError(f"binding rejected: {errors[0]['error']}") _save(root, data, style) return resolve_fabric_profile(root, model, task_type) def _profile_map(root): data, _ = _load(root) configuration = data.get("configuration") or {} if not isinstance(configuration, dict): raise ValueError("configuration must be an object") custom = configuration.get("custom_profiles") or {} if not isinstance(custom, dict): raise ValueError("configuration.custom_profiles must be an object") merged = {name: {**copy.deepcopy(profile), "builtin": True} for name, profile in BUILTIN_PROFILES.items()} for name, profile in custom.items(): if not _valid_profile_name(name): raise ValueError(f"invalid custom profile name {name!r}") if name in merged: raise ValueError(f"custom profile {name!r} shadows a built-in profile") if not isinstance(profile, dict): raise ValueError(f"custom profile {name!r} must be an object") merged[name] = {**copy.deepcopy(profile), "builtin": False} return merged def list_profiles(root="."): return _profile_map(root) def show_profile(root, name): profiles = _profile_map(root) if name not in profiles: raise KeyError(f"unknown profile {name!r}") result = copy.deepcopy(profiles[name]) result["name"] = name return result def _validated_profile_settings(root, name): profile = show_profile(root, name) settings = profile.get("settings") if not isinstance(settings, dict) or not settings: raise ValueError(f"profile {name!r} must contain non-empty settings") catalog = build_catalog(root) validated = {} for key, value in settings.items(): spec = catalog.get(key) if spec is None: raise ValueError(f"profile {name!r} contains unknown setting {key!r}") if not spec["mutable"]: raise ValueError(f"profile {name!r} contains read-only setting {key!r}") validated[key] = _coerce(spec, value) return profile, validated def diff_profile(root, name): profile, settings = _validated_profile_settings(root, name) changes = [] for key, desired in settings.items(): current = get(root, key) if current != desired: changes.append({"key": key, "current": current, "profile_value": desired}) return {"profile": name, "description": profile.get("description", ""), "changes": changes} def apply_profile(root, name, *, dry_run=False): profile, settings = _validated_profile_settings(root, name) data, style = _load(root) updated = copy.deepcopy(data) changes = [] for key, desired in settings.items(): current, present = _dig(data, key) if not present: current = build_catalog(root)[key]["default"] if current != desired: changes.append({"key": key, "from": current, "to": desired}) _put(updated, key, copy.deepcopy(desired)) _put(updated, "configuration.active_profile", name) if not dry_run: _save(root, updated, style) return { "profile": name, "description": profile.get("description", ""), "dry_run": dry_run, "changes": changes, } def _valid_profile_name(name): return isinstance(name, str) and re.fullmatch(r"[a-z][a-z0-9_-]{0,63}", name) is not None def save_profile(root, name, *, keys=None, description=""): if not _valid_profile_name(name): raise ValueError("profile name must start with a lowercase letter and contain only lowercase letters, numbers, _ or -") if name in BUILTIN_PROFILES: raise ValueError(f"cannot overwrite built-in profile {name!r}") catalog = build_catalog(root) selected = list(keys or [key for key, spec in catalog.items() if spec["mutable"] and spec["latency"] != "none"]) if not selected: raise ValueError("profile must capture at least one setting") settings = {} for key in selected: spec = catalog.get(key) if spec is None: raise KeyError(f"unknown setting {key!r}") if not spec["mutable"]: raise ValueError(f"cannot capture read-only setting {key!r}") settings[key] = get(root, key) data, style = _load(root) configuration = data.get("configuration") if configuration is None: configuration = {} data["configuration"] = configuration if not isinstance(configuration, dict): raise ValueError("configuration must be an object") profiles = configuration.setdefault("custom_profiles", {}) if not isinstance(profiles, dict): raise ValueError("configuration.custom_profiles must be an object") profiles[name] = {"description": description or f"Custom profile {name}", "settings": settings} configuration["active_profile"] = "custom" _save(root, data, style) return {"name": name, "description": profiles[name]["description"], "settings": settings} def delete_profile(root, name): if name in BUILTIN_PROFILES: raise ValueError(f"cannot delete built-in profile {name!r}") data, style = _load(root) configuration = data.get("configuration") if configuration is None: configuration = {} data["configuration"] = configuration if not isinstance(configuration, dict): raise ValueError("configuration must be an object") profiles = configuration.setdefault("custom_profiles", {}) if not isinstance(profiles, dict): raise ValueError("configuration.custom_profiles must be an object") deleted = profiles.pop(name, None) is not None if deleted: if configuration.get("active_profile") == name: configuration["active_profile"] = "custom" _save(root, data, style) return {"name": name, "deleted": deleted} def main(argv=None): parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--repo-root", default=".") parser.add_argument("--json", action="store_true") sub = parser.add_subparsers(dest="action") sub.required = False def json_flag(command): command.add_argument("--json", action="store_true", dest="json2") return command p_list = json_flag(sub.add_parser("list", help="List classified persistent settings.")) p_list.add_argument("--category") p_list.add_argument("--latency", choices=("high", "medium", "low", "none")) p_list.add_argument("--mutable", choices=("all", "yes", "no"), default="all") p_get = json_flag(sub.add_parser("get", help="Get one setting's value.")) p_get.add_argument("key") p_set = json_flag(sub.add_parser("set", help="Set one operator-mutable setting.")) p_set.add_argument("key") p_set.add_argument("value") for action in ("enable", "disable", "describe", "effective"): command = json_flag(sub.add_parser(action)) command.add_argument("key") fabric_cmd = json_flag(sub.add_parser( "set-fabric-binding", help="Mint one evidence-bound model->fabric-profile binding (fail-closed).")) fabric_cmd.add_argument("--model", required=True) fabric_cmd.add_argument("--profile", required=True) fabric_cmd.add_argument("--evidence", default=None) fabric_cmd.add_argument("--task-type", default=None) json_flag(sub.add_parser("audit", help="Validate config coverage, keys, and value types.")) json_flag(sub.add_parser( "journal-contract", help="Report whether the journal.* contract is valid, invalid (partial), or unavailable.")) json_flag(sub.add_parser("inventory", help="Report every persistent and non-persistent control plane.")) profile = sub.add_parser("profile", help="List, inspect, compare, apply, save, or delete profiles.") profile_sub = profile.add_subparsers(dest="profile_action", required=True) json_flag(profile_sub.add_parser("list")) for action in ("show", "diff", "delete"): command = json_flag(profile_sub.add_parser(action)) command.add_argument("name") apply_cmd = json_flag(profile_sub.add_parser("apply")) apply_cmd.add_argument("name") apply_cmd.add_argument("--dry-run", action="store_true") save_cmd = json_flag(profile_sub.add_parser("save")) save_cmd.add_argument("name") save_cmd.add_argument("--key", action="append", dest="keys") save_cmd.add_argument("--description", default="") args = parser.parse_args(argv) as_json = getattr(args, "json", False) or getattr(args, "json2", False) root = Path(args.repo_root) try: if args.action == "set": spec = build_catalog(root).get(args.key) raw = args.value if spec and spec["type"] in {"json_list", "json_object"}: raw = json.loads(raw) result = {"key": args.key, "value": set_toggle(root, args.key, raw)} elif args.action == "get": result = {"key": args.key, "value": get(root, args.key)} elif args.action == "enable": result = {"key": args.key, "value": enable(root, args.key)} elif args.action == "disable": result = {"key": args.key, "value": disable(root, args.key)} elif args.action == "describe": result = describe(root, args.key) elif args.action == "effective": result = effective(root, args.key) elif args.action == "set-fabric-binding": result = set_fabric_binding(root, args.model, args.profile, evidence=args.evidence, task_type=args.task_type) elif args.action == "audit": result = audit_config(root) elif args.action == "inventory": result = inventory(root) elif args.action == "journal-contract": result = journal_contract_state(root) elif args.action == "profile": if args.profile_action == "list": result = list_profiles(root) elif args.profile_action == "show": result = show_profile(root, args.name) elif args.profile_action == "diff": result = diff_profile(root, args.name) elif args.profile_action == "apply": result = apply_profile(root, args.name, dry_run=args.dry_run) elif args.profile_action == "save": result = save_profile(root, args.name, keys=args.keys, description=args.description) else: result = delete_profile(root, args.name) else: mutable = None if args.action is None or args.mutable == "all" else args.mutable == "yes" result = list_settings(root, category=getattr(args, "category", None), latency=getattr(args, "latency", None), mutable=mutable) if as_json: print(json.dumps(result, indent=2, ensure_ascii=False)) elif isinstance(result, list): for row in result: print(f"{row['key']} = {row['value']} [{row['latency']} latency; " f"{'mutable' if row['mutable'] else row['manager']}] -- {row['controls']}") elif args.action == "profile" and args.profile_action == "list": for name, row in result.items(): print(f"{name}: {row.get('description', '')}" + (" [built-in]" if row.get("builtin") else "")) else: print(json.dumps(result, indent=2, ensure_ascii=False)) except (json.JSONDecodeError, KeyError, ValueError) as exc: print(f"[PRD Plugin] error: {exc}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": sys.exit(main())