work_block_schema.py

# ---- Changelog ----
# [2026-05-03] Claude (Sonnet 4.6) — Add read_only_paths to constraints schema (#168)
# What: Added read_only_paths as a valid constraints property
# Why: spec_executor.py already reads this field (lines 96-100) but work_block_schema.py
#      constraints had additionalProperties:False with no read_only_paths entry — any spec
#      using it failed schema validation before execution could reach the field.
# How: One new property added to constraints; no executor changes (already handled).
# [2026-04-06] Josh + Claude — Add edit_file tool + shell_allowlist constraint
# What: (1) edit_file in _TOOL_NAMES (2) shell_allowlist in constraints schema
# Why: Gap 2 — specs need to extend shell allowlist; Gap 3 — edit_file is a new tool
# How: New enum entry in _TOOL_NAMES, new array property in constraints
# [2026-04-05] Josh + Claude — Structured work block spec schema
# What: JSON Schema for QB → Codemine worker handoff
# Why: Structured specs produce +37% better agent execution vs prose (zero improvisation, 97% validation)
# How: jsonschema validation, mandatory validation blocks on every action, on_failure handlers
# -------------------

"""Work Block Spec schema and validation.

Every action step MUST have a validation block and an on_failure handler.
The format is the guardrail — not the model.
"""

import json
import logging
from typing import Tuple

from jsonschema import Draft202012Validator, ValidationError

logger = logging.getLogger("work_block_schema")

# Tools available in Codemine's TOOL_REGISTRY
_TOOL_NAMES = [
    "read_file", "write_file", "edit_file", "list_files",
    "search_code", "search_conversations", "search_testament",
    "ingest_workspace", "shell_execute",
    "push_to_github", "pull_from_github", "create_shadow_branch",
    "notebook_read", "notebook_add", "notebook_delete",
    "map_repository_structure", "get_stats",
]

_CONDITION_OPERATORS = [
    "contains", "not_contains",
    "equals", "not_equals",
    "matches_regex",
    "result_is_string",
    "result_is_not_error",
    "file_exists",
    "file_contains",
    "output_length_gt", "output_length_lt",
]

_FAILURE_ACTIONS = ["abort_block", "retry", "skip", "goto", "escalate_to_qb"]

_GATE_TYPES = ["human_review", "qb_checkpoint", "auto_approve"]

# ---------------------------------------------------------------------------
# Sub-schemas (referenced by $defs in the main schema)
# ---------------------------------------------------------------------------

_CONDITION_CHECK = {
    "type": "object",
    "required": ["operator"],
    "additionalProperties": False,
    "properties": {
        "operator": {"enum": _CONDITION_OPERATORS},
        "target": {"type": "string"},
        "value": {},  # any type — depends on operator
        "description": {"type": "string"},
    },
}

# String shorthands accepted in on_failure for easier spec authoring.
# Executor normalizes these to the full object form before acting on them.
_FAILURE_SHORTHANDS = ["abort", "abort_block", "continue", "skip", "retry", "goto", "escalate_to_qb"]

_FAILURE_HANDLER = {
    "oneOf": [
        {
            "type": "string",
            "enum": _FAILURE_SHORTHANDS,
        },
        {
            "type": "object",
            "required": ["action"],
            "additionalProperties": False,
            "properties": {
                "action": {"enum": _FAILURE_ACTIONS},
                "max_retries": {"type": "integer", "minimum": 0, "default": 0},
                "goto_step": {"type": "string"},
                "message": {"type": "string"},
            },
        },
    ]
}

_VALIDATION_BLOCK = {
    "type": "object",
    "required": ["checks"],
    "additionalProperties": False,
    "properties": {
        "checks": {
            "type": "array",
            "items": _CONDITION_CHECK,
            "minItems": 1,
        },
    },
}

# Step definitions — each is a separate schema, unified via oneOf in the main schema
_ACTION_STEP = {
    "type": "object",
    "required": ["id", "type", "tool", "params", "validation", "on_failure"],
    "additionalProperties": False,
    "properties": {
        "id": {"type": "string"},
        "type": {"const": "action"},
        "description": {"type": "string"},
        "tool": {"enum": _TOOL_NAMES},
        "params": {"type": "object"},
        "bind_result": {"type": "string", "pattern": r"^\$[a-z_][a-z0-9_]*$"},
        "validation": _VALIDATION_BLOCK,
        "on_failure": _FAILURE_HANDLER,
    },
}

_GATE_STEP = {
    "type": "object",
    "required": ["id", "type", "description", "gate_type"],
    "additionalProperties": False,
    "properties": {
        "id": {"type": "string"},
        "type": {"const": "gate"},
        "description": {"type": "string"},
        "gate_type": {"enum": _GATE_TYPES},
        "staged_actions": {"type": "array", "items": {"type": "string"}},
        "timeout_seconds": {"type": "integer", "default": 300},
        "on_timeout": {"enum": ["abort", "skip", "auto_approve"], "default": "abort"},
    },
}

# Forward-reference placeholder — condition, loop, group contain nested steps.
# jsonschema handles recursive $ref via the $defs mechanism, but since we're
# building the schema as a Python dict, we use a sentinel and patch it below.
_STEP_REF = {"$ref": "#/$defs/step"}

_CONDITION_STEP = {
    "type": "object",
    "required": ["id", "type", "check", "if_true", "if_false"],
    "additionalProperties": False,
    "properties": {
        "id": {"type": "string"},
        "type": {"const": "condition"},
        "description": {"type": "string"},
        "check": _CONDITION_CHECK,
        "if_true": {"type": "array", "items": _STEP_REF},
        "if_false": {"type": "array", "items": _STEP_REF},
    },
}

_LOOP_STEP = {
    "type": "object",
    "required": ["id", "type", "over", "body"],
    "additionalProperties": False,
    "properties": {
        "id": {"type": "string"},
        "type": {"const": "loop"},
        "description": {"type": "string"},
        "over": {
            "oneOf": [
                {
                    "type": "object",
                    "required": ["items"],
                    "additionalProperties": False,
                    "properties": {
                        "items": {"type": "array", "items": {"type": "string"}},
                    },
                },
                {
                    "type": "object",
                    "required": ["from_result"],
                    "additionalProperties": False,
                    "properties": {
                        "from_result": {"type": "string"},
                        "split_on": {"type": "string", "default": "\n"},
                    },
                },
            ],
        },
        "bind_item": {"type": "string", "pattern": r"^\$[a-z_][a-z0-9_]*$", "default": "$item"},
        "max_iterations": {"type": "integer", "default": 20},
        "body": {"type": "array", "items": _STEP_REF, "minItems": 1},
    },
}

_GROUP_STEP = {
    "type": "object",
    "required": ["id", "type", "steps"],
    "additionalProperties": False,
    "properties": {
        "id": {"type": "string"},
        "type": {"const": "group"},
        "description": {"type": "string"},
        "steps": {"type": "array", "items": _STEP_REF, "minItems": 1},
        "on_failure": _FAILURE_HANDLER,
    },
}

# ---------------------------------------------------------------------------
# Main schema
# ---------------------------------------------------------------------------

WORK_BLOCK_SCHEMA = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "title": "WorkBlockSpec",
    "description": "Structured execution spec for QB -> Codemine worker handoff.",
    "type": "object",
    "required": ["spec_version", "block", "steps", "constraints"],
    "additionalProperties": False,
    "properties": {
        "spec_version": {"const": "1.0.0"},
        "block": {
            "type": "object",
            "required": ["id", "name", "scope", "acceptance_criteria"],
            "additionalProperties": False,
            "properties": {
                "id": {"type": "string"},
                "name": {"type": "string", "maxLength": 120},
                "agent": {"type": "string"},
                "scope": {"type": "string"},
                "workspace": {
                    "type": "string",
                    "description": "Workspace root for PolicyEngine path checks. Defaults to Codemine repo if omitted.",
                },
                "acceptance_criteria": {
                    "type": "array",
                    "items": {"type": "string"},
                    "minItems": 1,
                },
                "depends_on": {
                    "type": "array",
                    "items": {"type": "string"},
                    "default": [],
                },
            },
        },
        "snap_interface": {
            "type": "object",
            "additionalProperties": False,
            "properties": {
                "inputs": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "required": ["name", "type", "source_block"],
                        "additionalProperties": False,
                        "properties": {
                            "name": {"type": "string"},
                            "type": {"enum": ["file", "function", "config", "state"]},
                            "source_block": {"type": "string"},
                            "path": {"type": "string"},
                        },
                    },
                    "default": [],
                },
                "outputs": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "required": ["name", "type", "path"],
                        "additionalProperties": False,
                        "properties": {
                            "name": {"type": "string"},
                            "type": {"enum": ["file", "function", "config", "state"]},
                            "path": {"type": "string"},
                            "contract": {"type": "string"},
                        },
                    },
                    "default": [],
                },
            },
        },
        "constraints": {
            "type": "object",
            "required": ["never", "anti_drift"],
            "additionalProperties": False,
            "properties": {
                "never": {"type": "array", "items": {"type": "string"}, "minItems": 1},
                "anti_drift": {"type": "array", "items": {"type": "string"}},
                "tool_allowlist": {"type": "array", "items": {"enum": _TOOL_NAMES}},
                "shell_allowlist": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Additional shell command prefixes allowed for this spec, extends PolicyEngine's base allowlist.",
                },
                "max_iterations": {"type": "integer", "minimum": 1, "maximum": 100, "default": 15},
                "timeout_seconds": {"type": "integer", "minimum": 30, "maximum": 3600, "default": 300},
                "read_only_paths": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Absolute paths outside workspace that read_file may access (read-only). Typical use: sibling repos QB needs to inspect without writing.",
                },
            },
        },
        "steps": {
            "type": "array",
            "items": _STEP_REF,
            "minItems": 1,
        },
    },
    "$defs": {
        "step": {
            "oneOf": [
                _ACTION_STEP,
                _GATE_STEP,
                _CONDITION_STEP,
                _LOOP_STEP,
                _GROUP_STEP,
            ],
        },
    },
}

# Pre-compile the validator for reuse
_validator = Draft202012Validator(WORK_BLOCK_SCHEMA)


def validate_spec(spec: dict) -> Tuple[bool, list]:
    """Validate a work block spec against the schema.

    Returns (is_valid, errors) where errors is a list of human-readable strings.
    """
    errors = []
    for error in sorted(_validator.iter_errors(spec), key=lambda e: list(e.path)):
        path = ".".join(str(p) for p in error.absolute_path) or "(root)"
        errors.append(f"{path}: {error.message}")

    if errors:
        logger.warning("Spec validation failed with %d errors", len(errors))
    return len(errors) == 0, errors


def validate_spec_file(path: str) -> Tuple[bool, list]:
    """Load and validate a JSON spec file."""
    try:
        with open(path, "r") as f:
            spec = json.load(f)
    except (json.JSONDecodeError, OSError) as e:
        return False, [f"Failed to load spec: {e}"]
    return validate_spec(spec)