feat(grammar): use JSON Schema for typed outputs instead of a bespoke DSL

Replace the hand-rolled `outputs` type mini-language with standard JSON Schema
(Draft 2020-12), validated by the already-vendored `jsonschema` library. The
`outputs` field is now an inline JSON Schema, giving enums, numeric/string
constraints, unions, objects with dynamic keys, `$ref` reuse, and top-level
non-object contracts, and it deletes the custom schema compiler and its
reserved-key parsing footguns.

Validation is strict and does not coerce: a produced value whose types do not
already match the contract is a failure (a violation is a hard failure for a
single-model task, and a failed branch under the completion policy for a
multi-model task), which surfaces malformed model output rather than silently
reshaping it. The schema is checked for well-formedness at load time.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: anticomputer

doc/GRAMMAR.md

@@ -409,9 +409,11 @@ Three fields drive this:
 
 - `id` names a task. Its produced value (its final tool result) is exposed to
   later tasks as `outputs.<id>`.
-- `outputs` declares an inline schema. When present, the task's value is
-  validated and coerced against it before being stored. A malformed schema is
-  rejected when the taskflow is loaded, before any model calls are made.
+- `outputs` declares an inline JSON Schema (Draft 2020-12). When present, the
+  task's value is validated against it before being stored. Validation is strict
+  and does not coerce, so a value whose types do not match the contract is a
+  failure. A malformed schema is rejected when the taskflow is loaded, before
+  any model calls are made.
 - `over` is an explicit iterable selector for `repeat_prompt`: a Jinja
   expression evaluated against the template context (so it yields a real list,
   not a re-parsed string).
@@ -425,11 +427,17 @@ Example: one task produces a typed list of functions, the next analyses each.
       user_prompt: |
         List all functions as JSON: {"functions": [{"name": ..., "body": ...}]}
       outputs:
-        functions:
-          type: list
-          items:
-            name: str
-            body: str
+        type: object
+        properties:
+          functions:
+            type: array
+            items:
+              type: object
+              properties:
+                name: {type: string}
+                body: {type: string}
+              required: [name, body]
+        required: [functions]
   - task:
       repeat_prompt: true
       over: "outputs.list_functions.functions"
@@ -439,17 +447,20 @@ Example: one task produces a typed list of functions, the next analyses each.
         {{ result.body }}
 ```
 
-The `outputs` schema is a mapping of field name to type spec:
+The `outputs` schema is a standard JSON Schema (Draft 2020-12), authored inline
+in YAML, so the full vocabulary is available:
 
-- Scalars (as strings): `str`, `int`, `float`, `bool`, `any` (plus synonyms
-  `string`/`integer`/`number`/`boolean`). A trailing `?` marks a field
-  optional, e.g. `note: str?`.
-- Lists (as strings): `list` (list of any) or `list[T]` for a scalar `T`, e.g.
-  `tags: list[str]`.
-- Nested objects: a mapping whose keys are sub-field specs, or the explicit
-  `{type: object, fields: {...}}` form.
-- Lists of objects: `{type: list, items: <spec>}`, where `items` is any spec
-  (a scalar, a nested object mapping, etc.).
+- Types via `type` (`object`, `array`, `string`, `integer`, `number`,
+  `boolean`, `null`), with `properties`/`required` for objects and `items` for
+  arrays.
+- `enum`/`const` for fixed value sets, and constraints such as `minimum`,
+  `maximum`, `minLength`, and `pattern`.
+- Objects with dynamic keys via `additionalProperties`, and strictness via
+  `additionalProperties: false`.
+- Unions (`anyOf`/`oneOf`), and `$ref`/`$defs` to reuse a shape across fields.
+
+Because validation is strict (no coercion), have the task emit JSON whose types
+already match, e.g. the integer `7` rather than the string `"7"`.
 
 Notes:
 

examples/taskflows/example_pipeline.yaml

@@ -23,7 +23,13 @@ taskflow:
       run: |
         echo '{"findings": ["use-after-free in parser", "overflow in loader"]}'
       outputs:
-        findings: list[str]
+        type: object
+        properties:
+          findings:
+            type: array
+            items:
+              type: string
+        required: [findings]
   - task:
       # only triage when the audit found enough findings
       if: "outputs.audit.findings | length >= globals.min_findings"

examples/taskflows/example_typed_outputs.yaml

@@ -17,10 +17,16 @@ taskflow:
       # a shell task produces the structured object deterministically
       run: |
         echo '{"items": ["apples", "bananas", "oranges"]}'
-      # validate/coerce the produced value against this schema before it is
+      # validate the produced value against this JSON Schema before it is
       # stored as outputs.fruit_list
       outputs:
-        items: list[str]
+        type: object
+        properties:
+          items:
+            type: array
+            items:
+              type: string
+        required: [items]
   - task:
       repeat_prompt: true
       # iterate the typed output by name (no double-JSON heuristic)

src/seclab_taskflow_agent/models.py

@@ -201,10 +201,10 @@ def _validate_outputs(self) -> TaskDefinition:
         # Fail fast on a malformed outputs schema at load time, before any
         # model calls are made.
         if self.outputs:
-            from .output_schema import OutputSchemaError, build_output_model
+            from .output_schema import OutputSchemaError, validate_output_schema
 
             try:
-                build_output_model(f"{self.id or self.name or 'task'}_output", self.outputs)
+                validate_output_schema(self.outputs)
             except OutputSchemaError as exc:
                 raise ValueError(f"invalid 'outputs' schema: {exc}") from exc
         if self.over and not self.repeat_prompt:

src/seclab_taskflow_agent/output_schema.py

@@ -1,141 +1,77 @@
 # SPDX-FileCopyrightText: GitHub, Inc.
 # SPDX-License-Identifier: MIT
 
-"""Compile inline ``outputs`` schemas into Pydantic models.
+"""Validate task outputs against an inline JSON Schema.
 
-A task can declare a typed output contract inline:
+A task can declare a typed output contract inline as a JSON Schema (Draft
+2020-12), authored directly in YAML:
 
 .. code-block:: yaml
 
     outputs:
-      functions:
-        type: list
-        items:
-          name: str
-          body: str
-
-This module turns such a schema into a generated Pydantic model so a task's
-produced value can be validated and coerced before it is passed to downstream
-tasks as ``outputs.<id>``.
-
-Supported field type specs:
-
-* Scalars (as strings): ``str``, ``int``, ``float``, ``bool``, ``any`` (plus
-  the synonyms ``string``/``integer``/``number``/``boolean``). A trailing
-  ``?`` marks the field optional, e.g. ``str?``.
-* Lists (as strings): ``list`` (list of any) or ``list[T]`` where ``T`` is a
-  scalar type name, e.g. ``list[str]``.
-* Nested objects (as a mapping): a mapping whose keys are sub-field specs, or
-  the explicit ``{type: object, fields: {...}}`` form.
-* Lists of objects (as a mapping): ``{type: list, items: <spec>}`` where
-  ``items`` is any spec (a scalar string, a nested object mapping, etc.).
+      type: object
+      properties:
+        findings:
+          type: array
+          items:
+            type: object
+            properties:
+              file: {type: string}
+              severity: {type: string, enum: [low, medium, high]}
+            required: [file, severity]
+      required: [findings]
+
+The task's produced value is validated against the schema before it is passed
+to downstream tasks as ``outputs.<id>``. Using JSON Schema (validated with the
+``jsonschema`` library) rather than a bespoke type language gives enums,
+numeric/string constraints, unions, objects with dynamic keys, and ``$ref``
+reuse for free, and the schema itself is checked for well-formedness when the
+taskflow loads.
 """
 
 from __future__ import annotations
 
-__all__ = ["OutputSchemaError", "build_output_model", "validate_output"]
+__all__ = ["OutputSchemaError", "validate_output", "validate_output_schema"]
 
-from typing import Any, Optional
+from typing import Any
 
-from pydantic import BaseModel, ConfigDict, create_model
-
-_SCALARS: dict[str, Any] = {
-    "str": str,
-    "string": str,
-    "int": int,
-    "integer": int,
-    "float": float,
-    "number": float,
-    "bool": bool,
-    "boolean": bool,
-    "any": Any,
-}
+from jsonschema import Draft202012Validator
+from jsonschema.exceptions import SchemaError
 
 
 class OutputSchemaError(ValueError):
-    """Raised when an ``outputs`` schema is malformed or unsupported."""
-
-
-def _scalar_type(name: str, field: str) -> Any:
-    key = name.strip().lower()
-    if key not in _SCALARS:
-        raise OutputSchemaError(f"unsupported type {name!r} in field {field!r}")
-    return _SCALARS[key]
-
-
-def _spec_to_type(field: str, spec: Any, model_name: str) -> tuple[Any, bool]:
-    """Resolve a field spec into ``(python_type, optional)``."""
-    if isinstance(spec, str):
-        s = spec.strip()
-        optional = s.endswith("?")
-        if optional:
-            s = s[:-1].strip()
-        low = s.lower()
-        if low == "list":
-            return list, optional
-        if low.startswith("list[") and low.endswith("]"):
-            inner = s[5:-1].strip()
-            return list[_scalar_type(inner, field)], optional
-        return _scalar_type(s, field), optional
-
-    if isinstance(spec, dict):
-        optional = bool(spec.get("optional", False))
-        declared = spec.get("type")
-        if declared is None:
-            # Nested object shorthand: the mapping's keys are sub-fields.
-            return _build_model(f"{model_name}_{field}", spec), optional
-        declared = str(declared).lower()
-        if declared == "list":
-            items = spec.get("items", "any")
-            inner_type, _ = _spec_to_type(field, items, f"{model_name}_{field}")
-            return list[inner_type], optional
-        if declared == "object":
-            fields = spec.get("fields")
-            if not isinstance(fields, dict):
-                raise OutputSchemaError(
-                    f"object field {field!r} requires a 'fields' mapping"
-                )
-            return _build_model(f"{model_name}_{field}", fields), optional
-        # Scalar declared via the ``type`` key, e.g. {type: str}.
-        return _scalar_type(declared, field), optional
-
-    raise OutputSchemaError(f"invalid type spec for field {field!r}: {spec!r}")
-
-
-def _build_model(name: str, fields: Any) -> type[BaseModel]:
-    if not isinstance(fields, dict) or not fields:
-        raise OutputSchemaError(f"schema {name!r} must be a non-empty mapping of fields")
-    field_defs: dict[str, Any] = {}
-    for fname, fspec in fields.items():
-        if not isinstance(fname, str) or not fname.isidentifier():
-            raise OutputSchemaError(f"invalid field name {fname!r} in schema {name!r}")
-        py_type, optional = _spec_to_type(fname, fspec, name)
-        if optional:
-            field_defs[fname] = (Optional[py_type], None)
-        else:
-            field_defs[fname] = (py_type, ...)
-    return create_model(name, __config__=ConfigDict(extra="allow"), **field_defs)
-
-
-def build_output_model(name: str, schema: dict[str, Any]) -> type[BaseModel]:
-    """Compile an ``outputs`` schema mapping into a Pydantic model class.
+    """Raised when an ``outputs`` schema is not a well-formed JSON Schema."""
+
+
+def validate_output_schema(schema: Any) -> None:
+    """Check that *schema* is a well-formed JSON Schema, raising on failure.
+
+    Called at taskflow load time so a malformed output contract fails fast,
+    before any model calls are made.
 
     Raises:
-        OutputSchemaError: If the schema is malformed or uses an unsupported
-            type. Callers can use this to validate schemas at load time.
+        OutputSchemaError: If *schema* is not a non-empty, valid JSON Schema.
     """
-    return _build_model(name, schema)
+    if not isinstance(schema, dict) or not schema:
+        raise OutputSchemaError("'outputs' must be a non-empty JSON Schema object")
+    try:
+        Draft202012Validator.check_schema(schema)
+    except SchemaError as exc:
+        raise OutputSchemaError(f"invalid 'outputs' schema: {exc.message}") from exc
+
 
+def validate_output(schema: dict[str, Any], value: Any) -> Any:
+    """Validate *value* against the JSON Schema *schema* and return it unchanged.
 
-def validate_output(
-    schema: dict[str, Any], value: Any, model_name: str = "TaskOutput"
-) -> dict[str, Any]:
-    """Validate/coerce *value* against *schema*, returning a plain dict.
+    JSON Schema validation does not coerce, so a value whose types do not
+    already match the contract is a failure. This is deliberately stricter than
+    permissive coercion: it surfaces malformed model output instead of silently
+    reshaping it.
 
     Raises:
-        OutputSchemaError: If the schema itself is invalid.
-        pydantic.ValidationError: If *value* does not satisfy the schema.
+        OutputSchemaError: If *schema* itself is malformed.
+        jsonschema.ValidationError: If *value* does not satisfy *schema*.
     """
-    model = build_output_model(model_name, schema)
-    obj = model.model_validate(value)
-    return obj.model_dump()
+    validate_output_schema(schema)
+    Draft202012Validator(schema).validate(value)
+    return value

src/seclab_taskflow_agent/runner.py

@@ -28,7 +28,7 @@
 from typing import TYPE_CHECKING, Any
 
 import jinja2
-from pydantic import ValidationError
+from jsonschema.exceptions import ValidationError
 
 from ._stream import drive_backend_stream
 from ._watchdog import start_watchdog, watchdog_ping
@@ -272,13 +272,12 @@ def _aggregate_fanin(branches: list[_Branch]) -> list[dict[str, Any]]:
 def _capture_branch_output(
     branch: _Branch,
     schema: dict[str, Any],
-    output_id: str,
     *,
     agent_ok: bool,
 ) -> bool:
     """Validate one multi-model branch's result against the task's outputs schema.
 
-    Stores the coerced value on ``branch.output`` (``None`` when the branch
+    Stores the validated value on ``branch.output`` (``None`` when the branch
     failed to run or its result violates the contract) and returns whether the
     branch both ran and produced a schema-valid result. A contract violation is
     therefore surfaced as a failed branch, which the task's ``completion`` policy
@@ -299,7 +298,7 @@ def _capture_branch_output(
     except ValueError:
         value = branch.sink[-1].text
     try:
-        branch.output = validate_output(schema, value, model_name=f"{output_id}_output")
+        branch.output = validate_output(schema, value)
         return True
     except (ValidationError, ValueError) as exc:
         logging.error(
@@ -392,7 +391,7 @@ def _capture_task_output(
 
     if schema:
         value = decode_tool_result(last)
-        validated = validate_output(schema, value, model_name=f"{output_id}_output")
+        validated = validate_output(schema, value)
         store.set_output(output_id, validated)
     else:
         try:
@@ -1158,12 +1157,11 @@ async def _branch_record(tr: ToolResult) -> None:
                         # result against the task's `outputs` schema. A missing
                         # or schema-violating result makes it a failed branch,
                         # folded into the completion policy alongside run
-                        # failures; the coerced value feeds fan-in.
+                        # failures; the validated value feeds fan-in.
                         if multi_model and task_output_schema:
                             return _capture_branch_output(
                                 branch,
                                 task_output_schema,
-                                task_output_id or task_name,
                                 agent_ok=branch_ok,
                             )
                         return branch_ok