diff --git a/app/config/schemas/baseworkflow.yml b/app/config/schemas/baseworkflow.yml new file mode 100644 index 0000000..0f9bb06 --- /dev/null +++ b/app/config/schemas/baseworkflow.yml @@ -0,0 +1,179 @@ +# baseworkflow.yml — the declarative shelf-key contract (typed I/O). +# +# Each top-level key is a "shelf.key"; its value is the schema for what that key +# holds. A key is declared ONCE here and shared by every action that reads or +# writes it (input.job is read by 11 actions) — the DRY, single-source contract. +# The debug viewer reflects these into typed, validated probe forms; the engine +# validates written values against them when DISPATCH_ENFORCE_SCHEMA is set +# (opt-in, off by default). +# +# Field spec keys: type (string|integer|number|boolean|array|object|any), +# required, description, enum, items (array element type), format (e.g. +# multiline), default, example, minimum, maximum, fields (nested object). +# Pulled in by app/workflows/baseworkflow.yml via `schemas: [../config/schemas/*.yml]`. + +# -- run-time inputs (seeded onto the input shelf) -------------------------- +input.job: + type: object + description: The unit of work from intake — the GitHub issue (or synthesized job) the pipeline implements. + fields: + issue: {type: integer, required: true, description: "Issue number", example: 9001} + title: {type: string, required: true, description: "One-line summary", example: "Add retry with backoff to fetch"} + body: {type: string, format: multiline, description: "Full body / acceptance criteria", example: "Implement retry.\n\n## Acceptance criteria\n- retries 3x\n- exponential backoff"} + labels: {type: array, items: string, description: "GitHub labels (drive purpose classification)", example: [enhancement]} + discovered: {type: array, items: string, description: "Files surfaced during triage", example: ["foundation/proc.py", "foundation/runtime.py"]} + +input.triage: + type: object + description: The architect's routing decision for this job. + fields: + action: {type: string, required: true, enum: [implement, fix, docs], description: "Kind of work", example: implement} + scope: {type: string, enum: [s, m, l], description: "Rough size", example: m} + route: {type: string, description: "Generation route / model lane", example: gen-default} + confidence: {type: number, minimum: 0, maximum: 1, description: "Triage confidence (0–1)", example: 0.82} + +input.config: + type: object + description: Per-run pipeline configuration (defaulted when absent). + fields: + engineering_budget: {type: integer, minimum: 0, description: "Token budget for the engineering phase", example: 450000} + verify_cmd: {type: string, description: "Command that validates the produced work", example: "bash scripts/smoke.sh"} + +input.request: + type: object + description: The seed controller's raw intake envelope (ADR-003) — how the work entered. + fields: + source: {type: string, required: true, enum: [job, github], description: "Intake source", example: job} + job: {type: object, description: "Synthesized job (source=job): title / goal / acceptance"} + repo: {type: string, description: "owner/repo (source=github)"} + issue: {type: integer, description: "Issue number (source=github)"} + fields: {type: object, description: "Structured intake fields"} + +# -- deliverables read downstream (an action's inputs) ---------------------- +deliverables.purpose: + type: object + description: The classified purpose of the work — selects the engineering harness. + fields: + purpose: {type: string, description: "Classification", example: new-feature} + source: {type: string, enum: [label, keyword, default], description: "How it was classified"} + harness: {type: string, description: "Selected engineering harness/skill"} + harness_ref: {type: string, description: "Harness source ref"} + harness_setup: {type: string, format: multiline, description: "Setup commands"} + harness_rationale: {type: string, format: multiline, description: "Why this harness"} + +deliverables.plan: + type: object + description: The decomposed work plan — units plus agent staffing. + fields: + units: {type: array, items: object, description: "Work units (id, deliverable, files, phase, acceptance)"} + staffing: {type: object, description: "Agent staffing / wave plan"} + criteria: {type: array, description: "Acceptance criteria"} + +deliverables.strategy: + type: object + description: Parallelization strategy across the work units. + fields: + strategy: {type: string, description: "Chosen strategy"} + summary: {type: string, format: multiline} + requires_parallelization_rationale: {type: boolean} + parallelization_rationale: {type: string, format: multiline} + parallel_units: {type: array, items: string} + n_parallel: {type: integer, minimum: 0} + abc: {type: string} + +deliverables.bucket: + type: object + description: The budget bucket the work falls into. + fields: + bucket: {type: string, description: "Bucket label"} + k_units: {type: integer, minimum: 0, description: "Unit count"} + rationale: {type: string, format: multiline} + +deliverables.budget: + type: object + description: The engineering budget authorization envelope. + fields: + authorization: {type: object, description: "Session authorization (id, branch, route, budgets…)"} + k_units: {type: integer, minimum: 0} + engineering_budget: {type: integer, minimum: 0} + +deliverables.orchestration_script: + type: object + description: The compiled multi-agent orchestration script (foundation.actions.script). + fields: + phases: {type: array, items: object, description: "PhaseSpec list"} + permission: {type: string, description: "Permission posture", example: deny-by-default} + allow_tools: {type: array, items: string} + budget: {type: integer, minimum: 0} + +deliverables.work_plan: + type: object + description: The consolidated architect work plan handed to the engineer. + fields: + issue: {type: integer} + purpose: {type: object} + strategy: {type: object} + plan: {type: object} + budget: {type: object} + +deliverables.engineering_result: + type: object + description: The engineer's output for the work unit. + fields: + ok: {type: boolean, required: true, description: "Did the engineer succeed"} + value: {type: string, format: multiline, description: "Produced artifact / summary"} + meta: {type: object, description: "usage, model, source, budget"} + +deliverables.docs: + type: object + description: The docs update produced for the work. + fields: + issue: {type: integer} + summary: {type: string, format: multiline} + sections: {type: array, items: string} + +deliverables.consolidation: + type: object + description: The PR consolidation record (branch, PR, mutations). + fields: + issue: {type: string} + status: {type: string} + pr_number: {type: integer, description: "PR number (null until opened)"} + pr_url: {type: string} + branch: {type: string} + base: {type: string} + skipped: {type: boolean} + dry_run: {type: boolean} + mutations: {type: array, description: "gh mutations performed (or dry-run)"} + +# -- shared shelf (seed variant ↔ lifecycle hand-off, ADR-003) -------------- +shared.envelope: + type: object + description: The intake envelope passed between seed variants. + fields: + source: {type: string, enum: [job, github, interactive]} + repo: {type: string} + issue: {type: integer, description: "Issue number (null for job source)"} + fields: {type: object, description: "title / goal / acceptance"} + +shared.intake: + type: object + description: The seed controller's intake decision. + fields: + source: {type: string} + variant: {type: string, description: "Which seed variant handled it"} + accepted: {type: boolean, required: true} + unmet: {type: array, description: "Unmet needs"} + missing: {type: array, description: "Missing fields"} + reason: {type: string, format: multiline} + +shared.payload: + type: object + description: The normalized work payload the seed hands to spec. + fields: + title: {type: string, required: true} + goal: {type: string, format: multiline} + acceptance: {type: string, format: multiline} + labels: {type: array, items: string} + open_needs: {type: array, items: string} + provenance: {type: object, description: "kind + origin"} diff --git a/app/workflows/baseworkflow.yml b/app/workflows/baseworkflow.yml index 793dd21..a78610d 100644 --- a/app/workflows/baseworkflow.yml +++ b/app/workflows/baseworkflow.yml @@ -13,6 +13,12 @@ uses: # this file's directory). - ../config/actions/**/*.yml # #base-style include of every action interface file +# The declarative shelf-key contract: typed I/O for every "shelf.key" the actions +# read/write, authored once per key (schema.py). Drives typed debug-viewer probe +# forms and opt-in engine enforcement (DISPATCH_ENFORCE_SCHEMA). +schemas: + - ../config/schemas/*.yml + inputs: [request, job, triage] # provided at run time (seeded onto the input shelf) # The parties to the contract (ADR-003): controllers the workflow references — diff --git a/foundation/workflow/__init__.py b/foundation/workflow/__init__.py index 82d4005..e5e5854 100644 --- a/foundation/workflow/__init__.py +++ b/foundation/workflow/__init__.py @@ -28,7 +28,14 @@ from .controller import CompiledWorkflow, YamlController from .graph import GraphVisitor, workflow_to_graph -from .loader import SchemaError, load_workflow, parse_document +from .loader import SchemaError, load_schemas, load_workflow, parse_document +from .schema import ( + FieldSchema, + SchemaContractError, + ShelfSchema, + parse_schema, + parse_schemas, +) from .manifest import ( ActionManifest, IORef, @@ -74,6 +81,8 @@ "TokenRegistry", "ActionBinding", "ControllerSpec", "TokenError", "UnknownToken", # manifest "ActionManifest", "IORef", "ManifestError", "load_manifests", "manifest_from_dict", + # schema (the declarative shelf-key contract) + "ShelfSchema", "FieldSchema", "SchemaContractError", "parse_schema", "parse_schemas", "load_schemas", # nodes "Node", "WorkflowNode", "PhaseNode", "SequenceNode", "ParallelNode", "LoopNode", "ActionRefNode", "ControllerRefNode", diff --git a/foundation/workflow/graph.py b/foundation/workflow/graph.py index 463e354..4055614 100644 --- a/foundation/workflow/graph.py +++ b/foundation/workflow/graph.py @@ -203,6 +203,9 @@ def visit_workflow(self, node: Any) -> Dict[str, Any]: "budgets": dict(node.budgets or {}), "inputs": list(node.inputs or ()), "controllers": list(getattr(node, "controllers", ()) or ()), + # The declarative shelf-key contract, as JSON, keyed by "shelf.key" — the + # viewer reflects these into typed, validated probe forms. + "schemas": {ref: s.to_dict() for ref, s in (getattr(node, "schemas", {}) or {}).items()}, "nodes": self.nodes, "edges": self.edges, } diff --git a/foundation/workflow/loader.py b/foundation/workflow/loader.py index e84a08e..08e22ce 100644 --- a/foundation/workflow/loader.py +++ b/foundation/workflow/loader.py @@ -22,6 +22,7 @@ from foundation import filesys from .manifest import ActionManifest, load_manifests +from .schema import parse_schemas from .nodes import ( ActionRefNode, ControllerRefNode, @@ -175,6 +176,32 @@ def _apply_header(merged: WorkflowNode, doc: Dict[str, Any], base: WorkflowNode) ) +def load_schemas(base_dir: str, spec: Any) -> Dict[str, Any]: + """Resolve a workflow's ``schemas:`` declaration into ``{"shelf.key": ShelfSchema}``. + + ``spec`` may be an **inline map** (``ref -> schema``) or a **list of file globs** + (each file a ``ref -> schema`` map), resolved relative to ``base_dir`` — the + ``#base`` convention, exactly like ``uses:``. Later files/keys win on collision. + The declarative-file form is the DRY authoring model: a key like ``input.job`` is + declared once and shared by every action that reads it.""" + if not spec: + return {} + if isinstance(spec, dict): + return parse_schemas(spec, where="schemas") + import glob as _glob + + import yaml + + merged: Dict[str, Any] = {} + for g in spec: + pattern = g if os.path.isabs(g) else os.path.join(base_dir, g) + for path in sorted(_glob.glob(pattern, recursive=True)): + with open(path, encoding="utf-8") as fh: + d = yaml.safe_load(fh) or {} + merged.update(parse_schemas(d, where=os.path.relpath(path, base_dir))) + return merged + + def load_workflow(relpath: str) -> WorkflowNode: """Read + parse a workflow YAML into a ``WorkflowNode``. ``relpath`` resolves via :func:`_resolve_path`; ``uses:`` manifest globs resolve relative to the @@ -189,17 +216,19 @@ def load_workflow(relpath: str) -> WorkflowNode: doc = yaml.safe_load(fh) or {} wf_dir = os.path.dirname(os.path.abspath(path)) manifests = load_manifests(wf_dir, doc.get("uses") or []) + schemas = load_schemas(wf_dir, doc.get("schemas")) + + from dataclasses import replace extends = doc.get("extends") if extends: - from dataclasses import replace - from .overlay import apply_overlays base = load_workflow(os.path.join(wf_dir, f"{extends}.yml")) merged = apply_overlays(base, doc.get("overlays") or [], manifests, source=relpath) # The merged tree's interface library is the base's plus the overlay's own. merged = replace(merged, manifests={**base.manifests, **manifests}) - return _apply_header(merged, doc, base) + # Contract likewise extends the base's (overlay keys override same refs). + return replace(_apply_header(merged, doc, base), schemas={**base.schemas, **schemas}) - return parse_document(doc, manifests, source=relpath) + return replace(parse_document(doc, manifests, source=relpath), schemas=schemas) diff --git a/foundation/workflow/nodes.py b/foundation/workflow/nodes.py index 4d9e4d4..5ffb750 100644 --- a/foundation/workflow/nodes.py +++ b/foundation/workflow/nodes.py @@ -121,6 +121,11 @@ class WorkflowNode(Node): # declared-only controllers (dynamic supersede targets, never in a phase) # can still be expanded, needs-checked and materialized. manifests: Dict[str, Any] = field(default_factory=dict) + # The declared shelf-key contract (``"shelf.key" -> ShelfSchema``): typed I/O + # for the workflow, authored once per key (``schemas:`` include) and shared by + # every action that touches it. Drives typed debug-viewer forms and the + # opt-in engine enforcement (``DISPATCH_ENFORCE_SCHEMA``). + schemas: Dict[str, Any] = field(default_factory=dict) admin_spec_split: float = 0.5 # Optional declarative early-completion: a registered predicate name. When a # step completes *successfully* and this predicate admits its Result, the whole diff --git a/foundation/workflow/schema.py b/foundation/workflow/schema.py new file mode 100644 index 0000000..bcfb279 --- /dev/null +++ b/foundation/workflow/schema.py @@ -0,0 +1,270 @@ +#!/usr/bin/env python3 +"""schema.py — the declarative shelf-key contract (typed I/O for the workflow). + +An action's ``interface.in``/``out`` names *where* data lives (``alias -> shelf.key``); +this module names *what shape* a ``shelf.key`` holds. A shelf key like ``input.job`` +is read by many actions, so its type is a property of the **key**, not of any one +action — the contract is therefore a registry keyed by ``"shelf.key"``, declared +once (see ``app/config/schemas/*.yml``) and shared. + +The schema is a small, self-defined type spec — deliberately *not* JSON Schema, so +``foundation.workflow`` stays a leaf (no ``jsonschema`` runtime dependency, mirroring +the note in this package's ``__init__``). It carries exactly what a debug UI needs to +render a typed form and what the engine needs to check a value: + + type string | integer | number | boolean | array | object | any + required (object field) the key must be present + description human text (field label / tooltip) + enum allowed values (renders as a dropdown) + items element type for ``array`` + format rendering hint (e.g. ``multiline`` for a big string) + default value used when absent + example a realistic value (prefills the form) + minimum/maximum numeric bounds + fields nested field schemas for ``object`` + +``validate(value)`` returns a list of human-readable error strings (empty = ok) and +is shared by both the tooling surface (the debug viewer's probe validation) and the +opt-in engine enforcement (``DISPATCH_ENFORCE_SCHEMA``); ``to_dict()`` emits the +JSON-only shape a UI consumes. + +Leaf module: stdlib only (an in-function PyYAML import lives in the loader, not here). +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple + +SCALAR_TYPES = ("string", "integer", "number", "boolean") +VALID_TYPES = SCALAR_TYPES + ("array", "object", "any") + + +class SchemaError(Exception): + """A malformed schema declaration.""" + + +class SchemaContractError(Exception): + """A runtime shelf value violated its declared contract (raised only when + ``DISPATCH_ENFORCE_SCHEMA`` is set; off by default). Trapped by ``Action.run`` + into a failed Result, so a violation fails the step cleanly rather than + crashing the engine.""" + + +def _pytype(v: Any) -> str: + if isinstance(v, bool): + return "boolean" + if isinstance(v, int): + return "integer" + if isinstance(v, float): + return "number" + if isinstance(v, str): + return "string" + if isinstance(v, (list, tuple)): + return "array" + if isinstance(v, dict): + return "object" + if v is None: + return "null" + return type(v).__name__ + + +def _type_ok(value: Any, t: str) -> bool: + if t in ("any", ""): + return True + if t == "string": + return isinstance(value, str) + if t == "integer": + return isinstance(value, int) and not isinstance(value, bool) + if t == "number": + return isinstance(value, (int, float)) and not isinstance(value, bool) + if t == "boolean": + return isinstance(value, bool) + if t == "array": + return isinstance(value, (list, tuple)) + if t == "object": + return isinstance(value, dict) + return True + + +@dataclass(frozen=True) +class FieldSchema: + """One field of an ``object`` shelf schema (or one array element's contract).""" + + name: str + type: str = "any" + required: bool = False + description: str = "" + enum: Tuple[Any, ...] = () + items: str = "" + format: str = "" + default: Any = None + example: Any = None + minimum: Optional[float] = None + maximum: Optional[float] = None + fields: Tuple["FieldSchema", ...] = () + + def validate(self, value: Any) -> List[str]: + """Errors for ``value`` against this field (no key prefix; the caller adds it).""" + if value is None: + return [] # presence is the parent object's ``required`` concern + errs: List[str] = [] + if not _type_ok(value, self.type): + return [f"expected {self.type}, got {_pytype(value)}"] + if self.enum and value not in self.enum: + errs.append(f"{value!r} not in {list(self.enum)}") + if self.type in ("number", "integer"): + if self.minimum is not None and value < self.minimum: + errs.append(f"{value} < minimum {self.minimum}") + if self.maximum is not None and value > self.maximum: + errs.append(f"{value} > maximum {self.maximum}") + if self.type == "array" and self.items: + for i, item in enumerate(value): + if not _type_ok(item, self.items): + errs.append(f"[{i}] expected {self.items}, got {_pytype(item)}") + if self.type == "object" and self.fields: + errs.extend(_validate_object(self.fields, value)) + return errs + + def to_dict(self) -> Dict[str, Any]: + out: Dict[str, Any] = {"name": self.name, "type": self.type} + if self.required: + out["required"] = True + if self.description: + out["description"] = self.description + if self.enum: + out["enum"] = list(self.enum) + if self.items: + out["items"] = self.items + if self.format: + out["format"] = self.format + if self.default is not None: + out["default"] = self.default + if self.example is not None: + out["example"] = self.example + if self.minimum is not None: + out["minimum"] = self.minimum + if self.maximum is not None: + out["maximum"] = self.maximum + if self.fields: + out["fields"] = [f.to_dict() for f in self.fields] + return out + + +def _validate_object(fields: Tuple[FieldSchema, ...], value: Dict[str, Any]) -> List[str]: + errs: List[str] = [] + for f in fields: + present = f.name in value + if f.required and not present: + errs.append(f".{f.name}: required") + continue + if present: + errs.extend(f".{f.name}{'' if e[:1] in ('.', '[') else ': '}{e}" for e in f.validate(value[f.name])) + return errs + + +@dataclass(frozen=True) +class ShelfSchema: + """The declared contract for one ``shelf.key`` (the registry value).""" + + ref: str + type: str = "object" + description: str = "" + fields: Tuple[FieldSchema, ...] = () + items: str = "" + enum: Tuple[Any, ...] = () + example: Any = None + + def validate(self, value: Any) -> List[str]: + """Human-readable errors for ``value`` against this contract (empty = ok).""" + if value is None: + return [] + if not _type_ok(value, self.type): + return [f"{self.ref}: expected {self.type}, got {_pytype(value)}"] + errs: List[str] = [] + if self.type == "object": + errs.extend(f"{self.ref}{e}" for e in _validate_object(self.fields, value)) + elif self.type == "array" and self.items: + for i, item in enumerate(value): + if not _type_ok(item, self.items): + errs.append(f"{self.ref}[{i}]: expected {self.items}, got {_pytype(item)}") + if self.enum and value not in self.enum: + errs.append(f"{self.ref}: {value!r} not in {list(self.enum)}") + return errs + + def to_dict(self) -> Dict[str, Any]: + out: Dict[str, Any] = {"ref": self.ref, "type": self.type} + if self.description: + out["description"] = self.description + if self.fields: + out["fields"] = [f.to_dict() for f in self.fields] + if self.items: + out["items"] = self.items + if self.enum: + out["enum"] = list(self.enum) + if self.example is not None: + out["example"] = self.example + return out + + +# -- parsing ---------------------------------------------------------------- +def _as_field(name: str, spec: Any, where: str) -> FieldSchema: + # Shorthand: ``title: string`` (a bare type name) or ``title: {type: string, ...}``. + if isinstance(spec, str): + spec = {"type": spec} + if not isinstance(spec, dict): + raise SchemaError(f"{where}.{name}: field must be a type string or a map, got {type(spec).__name__}") + t = str(spec.get("type", "any")) + if t not in VALID_TYPES: + raise SchemaError(f"{where}.{name}: unknown type {t!r} (one of {VALID_TYPES})") + nested = spec.get("fields") or {} + return FieldSchema( + name=name, + type=t, + required=bool(spec.get("required", False)), + description=str(spec.get("description", "")), + enum=tuple(spec.get("enum") or ()), + items=str(spec.get("items", "")), + format=str(spec.get("format", "")), + default=spec.get("default"), + example=spec.get("example"), + minimum=spec.get("minimum"), + maximum=spec.get("maximum"), + fields=tuple(_as_field(k, v, f"{where}.{name}") for k, v in nested.items()), + ) + + +def parse_schema(ref: str, spec: Any) -> ShelfSchema: + """Parse one ``"shelf.key" -> spec`` declaration into a :class:`ShelfSchema`.""" + if isinstance(spec, str): + spec = {"type": spec} + if not isinstance(spec, dict): + raise SchemaError(f"schema {ref!r}: must be a map, got {type(spec).__name__}") + t = str(spec.get("type", "object")) + if t not in VALID_TYPES: + raise SchemaError(f"schema {ref!r}: unknown type {t!r} (one of {VALID_TYPES})") + fields = spec.get("fields") or {} + if not isinstance(fields, dict): + raise SchemaError(f"schema {ref!r}: 'fields' must be a map of name -> field spec") + return ShelfSchema( + ref=ref, + type=t, + description=str(spec.get("description", "")), + fields=tuple(_as_field(k, v, ref) for k, v in fields.items()), + items=str(spec.get("items", "")), + enum=tuple(spec.get("enum") or ()), + example=spec.get("example"), + ) + + +def parse_schemas(mapping: Any, *, where: str = "schemas") -> Dict[str, ShelfSchema]: + """Parse a ``{"shelf.key": spec, ...}`` mapping into a schema registry.""" + if mapping is None: + return {} + if not isinstance(mapping, dict): + raise SchemaError(f"{where}: must be a map of 'shelf.key' -> schema, got {type(mapping).__name__}") + out: Dict[str, ShelfSchema] = {} + for ref, spec in mapping.items(): + if "." not in str(ref): + raise SchemaError(f"{where}: key {ref!r} must be 'shelf.key'") + out[str(ref)] = parse_schema(str(ref), spec) + return out diff --git a/foundation/workflow/test_schema.py b/foundation/workflow/test_schema.py new file mode 100644 index 0000000..85a50a9 --- /dev/null +++ b/foundation/workflow/test_schema.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +"""test_schema.py — the declarative shelf-key contract (parse + validate + to_dict).""" +from __future__ import annotations + +import pytest + +from foundation.workflow.schema import ( + SchemaError, + ShelfSchema, + parse_schema, + parse_schemas, +) + +JOB_SPEC = { + "type": "object", + "description": "a job", + "fields": { + "issue": {"type": "integer", "required": True}, + "title": {"type": "string", "required": True}, + "labels": {"type": "array", "items": "string"}, + "confidence": {"type": "number", "minimum": 0, "maximum": 1}, + "action": {"type": "string", "enum": ["implement", "fix", "docs"]}, + }, +} + + +def test_parse_shape(): + s = parse_schema("input.job", JOB_SPEC) + assert s.ref == "input.job" + assert s.type == "object" + names = {f.name: f for f in s.fields} + assert names["issue"].required and names["issue"].type == "integer" + assert names["labels"].items == "string" + assert names["action"].enum == ("implement", "fix", "docs") + assert names["confidence"].minimum == 0 and names["confidence"].maximum == 1 + + +def test_validate_ok(): + s = parse_schema("input.job", JOB_SPEC) + assert s.validate({"issue": 1, "title": "x", "labels": ["a"], "confidence": 0.5, "action": "fix"}) == [] + # absent optional fields are fine; only declared-required keys must be present + assert s.validate({"issue": 1, "title": "x"}) == [] + # None short-circuits (a missing shelf value is not a violation on its own) + assert s.validate(None) == [] + + +def test_validate_required_and_type(): + s = parse_schema("input.job", JOB_SPEC) + assert any("issue" in e and "required" in e for e in s.validate({"title": "x"})) + assert any("expected integer" in e for e in s.validate({"issue": "nine", "title": "x"})) + + +def test_validate_enum_range_and_items(): + s = parse_schema("input.job", JOB_SPEC) + assert any("not in" in e for e in s.validate({"issue": 1, "title": "x", "action": "destroy"})) + assert any("maximum" in e for e in s.validate({"issue": 1, "title": "x", "confidence": 9})) + errs = s.validate({"issue": 1, "title": "x", "labels": [1, 2]}) + assert sum("expected string" in e for e in errs) == 2 + + +def test_shorthand_type_string(): + # ``title: string`` shorthand and a scalar top-level schema. + reg = parse_schemas({"shared.name": "string"}) + assert reg["shared.name"].type == "string" + assert reg["shared.name"].validate("hi") == [] + assert reg["shared.name"].validate(3) == ["shared.name: expected string, got integer"] + + +def test_to_dict_roundtrips_key_fields(): + d = parse_schema("input.job", JOB_SPEC).to_dict() + assert d["ref"] == "input.job" and d["type"] == "object" + issue = next(f for f in d["fields"] if f["name"] == "issue") + assert issue["required"] is True and issue["type"] == "integer" + + +def test_bad_declarations_raise(): + with pytest.raises(SchemaError): + parse_schema("input.job", {"type": "widget"}) # unknown type + with pytest.raises(SchemaError): + parse_schemas({"noshelf": {"type": "object"}}) # key not "shelf.key" + with pytest.raises(SchemaError): + parse_schema("input.job", {"fields": {"x": {"type": "nope"}}}) # unknown field type + + +def test_empty_and_none(): + assert parse_schemas(None) == {} + assert ShelfSchema(ref="x.y").validate({}) == [] diff --git a/foundation/workflow/visitor.py b/foundation/workflow/visitor.py index 6d749ef..0fb6e7d 100644 --- a/foundation/workflow/visitor.py +++ b/foundation/workflow/visitor.py @@ -32,6 +32,7 @@ """ from __future__ import annotations +import os from dataclasses import dataclass from typing import Any, Callable, Dict, Iterable, List, Optional, Set @@ -42,6 +43,7 @@ from .loader import SchemaError, _parse_step from .nodes import ActionRefNode, ControllerRefNode, SequenceNode from .predicate import PredicateError, PredicateValidator, compile_predicate, parse_predicate +from .schema import SchemaContractError # Per-inference deterministic mock cost + default route — preserve the old # base workflow values so the budget meters and spend are identical. @@ -150,12 +152,24 @@ def _read_inputs(manifest: Any, ctx: Any) -> Dict[str, Any]: return inputs -def _write_outputs(manifest: Any, ctx: Any, out: Any) -> None: +def _write_outputs(manifest: Any, ctx: Any, out: Any, schemas: Optional[Dict[str, Any]] = None) -> None: if not isinstance(out, dict): return + # Staged enforcement (ADR: contract with teeth). Off by default; a run with + # ``DISPATCH_ENFORCE_SCHEMA`` set validates every written shelf value against + # its declared contract and fails the step on a violation. With no schemas + # wired (the common case) this is a no-op — zero cost to the live pipeline. + enforce = bool(schemas) and bool(os.environ.get("DISPATCH_ENFORCE_SCHEMA")) for ref in manifest.outputs: if ref.alias in out: - getattr(ctx.shelves, ref.shelf).put(ref.key, out[ref.alias]) + value = out[ref.alias] + if enforce: + schema = schemas.get(ref.ref) + if schema is not None: + errs = schema.validate(value) + if errs: + raise SchemaContractError(f"{manifest.token} wrote {ref.ref}: " + "; ".join(errs)) + getattr(ctx.shelves, ref.shelf).put(ref.key, value) # -- compile ---------------------------------------------------------------- @@ -166,11 +180,13 @@ def __init__(self, factory: Any, registry: Any) -> None: self.factory = factory self.registry = registry self.budgets: Dict[str, int] = {} + self.schemas: Dict[str, Any] = {} # declared shelf-key contract (for opt-in enforcement) self.terminal: Optional[Callable[[Any, Any], bool]] = None self._expanding: Set[str] = set() # cyclic controller-reference guard def visit_workflow(self, node: Any) -> CompiledWorkflow: self.budgets = dict(node.budgets or {}) + self.schemas = dict(getattr(node, "schemas", {}) or {}) # Compile the optional declarative early-completion predicate first, so the # per-phase Sequences built below can carry it (and the lifecycle too). tw = getattr(node, "terminal_when", None) @@ -270,11 +286,13 @@ def raw_body(payload: Any, ctx: Any) -> Any: return raw_body + schemas = getattr(self, "schemas", None) # captured for opt-in output enforcement + def body(payload: Any, ctx: Any) -> Any: inputs = _read_inputs(manifest, ctx) out = fn(inputs, ctx) if binding.needs_ctx else fn(inputs) if not manifest.raw: - _write_outputs(manifest, ctx, out) + _write_outputs(manifest, ctx, out, schemas=schemas) return out return body diff --git a/tools/debugviewer/server.py b/tools/debugviewer/server.py index 355f078..a7d031b 100644 --- a/tools/debugviewer/server.py +++ b/tools/debugviewer/server.py @@ -11,6 +11,9 @@ GET / -> the viewer (static/viewer.html) GET /api/workflows -> [{name, file, ok, error, phases, node_count}] GET /api/workflows/ -> the graph {name, phases, budgets, nodes, edges} + GET /api/workflows//examples + -> {ok, examples}: realistic "shelf.key"->value + snapshot from the offline mock, for typed probe forms GET /api/sessions -> {runs: [...], runnable: [...]} POST /api/probe -> run ONE action in isolation against the mock engine; body {workflow, token, inputs, payload} @@ -173,6 +176,18 @@ def probe_action(workflow: str, token: str, inputs: Dict[str, Any], payload: Any if manifest is None: return {"error": f"no action {token!r} in {workflow!r}"} + # Validate the seeded inputs against the declared contract (the tooling surface + # of the staged enforcement). Non-blocking: a probe still runs so you can see + # what a malformed value does, but the violations are reported to the UI. + schemas = getattr(doc, "schemas", {}) or {} + validation: Dict[str, List[str]] = {} + for ref, value in (inputs or {}).items(): + schema = schemas.get(str(ref)) + if schema is not None: + errs = schema.validate(value) + if errs: + validation[str(ref)] = errs + from foundation.actions import BudgetMeter, Context, MockActionFactory from foundation.workflow.nodes import ActionRefNode from foundation.workflow.visitor import CompileVisitor @@ -216,6 +231,7 @@ def probe_action(workflow: str, token: str, inputs: Dict[str, Any], payload: Any "ok": bool(result.ok), "token": token, "seeded": seeded, + "validation": validation, # {ref: [contract violations]} — empty when inputs conform "result": { "ok": bool(result.ok), "value": getattr(result, "value", None), @@ -229,6 +245,68 @@ def probe_action(workflow: str, token: str, inputs: Dict[str, Any], payload: Any }) +# -- example shelf snapshots (schema-driven probe forms) -------------------- +# Canonical, deterministic fixtures fed to a workflow's offline mock so we can +# read back *real* shelf values and reflect their shapes into typed form fields. +_EXAMPLE_JOB = { + "issue": 9001, + "title": "Add retry with backoff to fetch", + "body": "Implement retry.\n\n## Acceptance criteria\n- retries 3x\n- exponential backoff\n- gives up after cap", + "labels": ["enhancement"], + "discovered": ["foundation/proc.py", "foundation/runtime.py"], +} +_EXAMPLE_TRIAGE = {"action": "implement", "scope": "m", "route": "gen-default", "confidence": 0.82} + +# Workflows with a deterministic offline mock runner we can introspect (mirrors +# ``_REGISTRIES``). A draw-only workflow simply yields no examples and the probe +# UI falls back to freeform inputs. +_EXAMPLE_RUNNERS: Dict[str, str] = {"baseworkflow": "baseworkflow.baseworkflow:run_mock"} +_example_cache: Dict[str, Dict[str, Any]] = {} + + +def example_shelf(workflow: str) -> Dict[str, Any]: + """A realistic ``"shelf.key" -> value`` snapshot of every ref a workflow's + actions read/write, derived by running its offline mock once (no model, no + network) and reading the input + deliverables shelves. The probe UI reflects + these value shapes into typed, prefilled form controls. Best-effort and + cached: an un-runnable workflow returns ``{}`` (UI falls back to freeform).""" + if workflow in _example_cache: + return _example_cache[workflow] + spec = _EXAMPLE_RUNNERS.get(workflow) + examples: Dict[str, Any] = {} + if spec: + try: + import contextlib + import importlib + import io + + # Same sys.path discipline as ``_registry_for``: the package __init__ + # is the composition root, so REPO_ROOT must win, but baseworkflow/ + # is still needed for the package's bare-internal imports. + bw_dir = os.path.join(REPO_ROOT, "baseworkflow") + for p in (bw_dir, REPO_ROOT): + if p in sys.path: + sys.path.remove(p) + sys.path.insert(0, bw_dir) + sys.path.insert(0, REPO_ROOT) + os.environ["PIPELINE_DRY_RUN"] = "1" # the mock never mutates anything real + mod_name, _, fn = spec.partition(":") + run_mock = getattr(importlib.import_module(mod_name), fn or "run_mock") + with contextlib.redirect_stdout(io.StringIO()): # swallow the mock's DRY-RUN chatter + summary = run_mock(_EXAMPLE_JOB, _EXAMPLE_TRIAGE) + wf = summary.get("workflow") + shelves = getattr(wf, "shelves", None) + for kind in ("input", "deliverables", "shared"): + shelf = getattr(shelves, kind, None) + snap = shelf.snapshot() if shelf is not None and hasattr(shelf, "snapshot") else {} + for key, value in (snap or {}).items(): + examples[f"{kind}.{key}"] = _json_safe(value) + except Exception: # noqa: BLE001 — best-effort; the UI falls back to freeform inputs + examples = {} + _example_cache[workflow] = examples + return examples + + # -- HTTP handler ----------------------------------------------------------- class Handler(BaseHTTPRequestHandler): server_version = "dispatch-debugviewer/0.1" @@ -285,6 +363,10 @@ def do_GET(self) -> None: # noqa: N802 self._static("viewer.html", "text/html; charset=utf-8") elif route == "/api/workflows": self._json(list_workflows()) + elif route.startswith("/api/workflows/") and route.endswith("/examples"): + name = route[len("/api/workflows/"):-len("/examples")] + ex = example_shelf(name) + self._json({"ok": bool(ex), "examples": ex}) elif route.startswith("/api/workflows/"): name = route[len("/api/workflows/"):] graph = graph_for(name) diff --git a/tools/debugviewer/static/viewer.html b/tools/debugviewer/static/viewer.html index 3b3f693..473de99 100644 --- a/tools/debugviewer/static/viewer.html +++ b/tools/debugviewer/static/viewer.html @@ -4,21 +4,39 @@ dispatch · workflow debug viewer +