solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Offline CI gate for strict structured-output schema portability."""
5
6from __future__ import annotations
7
8import json
9from pathlib import Path
10from typing import Any
11
12import pytest
13
14from solstone.apps.timeline.rollup import build_rollup_schema
15from solstone.think.schema_prep import (
16 prepare_provider_schema,
17 unsupported_keyword_hits,
18)
19from solstone.think.talent import hydrate_runtime_enums
20
21REPO_ROOT = Path(__file__).resolve().parents[1]
22STRICT_PROVIDERS = ("openai", "anthropic", "google")
23
24
25def _discover_schemas() -> tuple[tuple[str, dict[str, Any]], ...]:
26 discovered: list[tuple[str, dict[str, Any]]] = []
27 for path in sorted((REPO_ROOT / "solstone").glob("**/*.schema.json")):
28 schema_id = path.relative_to(REPO_ROOT).as_posix()
29 schema = json.loads(path.read_text(encoding="utf-8"))
30 if isinstance(schema.get("x-journal-contract"), dict):
31 continue
32 discovered.append((schema_id, schema))
33 discovered.append(("build_rollup_schema(3)", build_rollup_schema(3)))
34 return tuple(discovered)
35
36
37SCHEMAS = _discover_schemas()
38
39
40def violations(schema: dict[str, Any]) -> list[str]:
41 found: list[str] = []
42
43 root_is_object = schema.get("type") == "object" or (
44 "properties" in schema and "type" not in schema
45 )
46 if not root_is_object:
47 found.append("$: root schema must be an object")
48
49 def walk(node: Any, path: str) -> None:
50 if isinstance(node, dict):
51 for key in node:
52 if key == "oneOf":
53 found.append(f"{path}: banned key 'oneOf'")
54
55 if node.get("type") == "object" or "properties" in node:
56 if node.get("additionalProperties") is not False:
57 found.append(f"{path}: object missing additionalProperties:false")
58 properties = node.get("properties") or {}
59 required = node.get("required") or []
60 missing = sorted(set(properties) - set(required))
61 if missing:
62 found.append(f"{path}: properties not required {missing!r}")
63
64 for key, value in node.items():
65 walk(value, f"{path}/{key}")
66 elif isinstance(node, list):
67 for index, value in enumerate(node):
68 walk(value, f"{path}[{index}]")
69
70 walk(schema, "$")
71 return found
72
73
74@pytest.mark.parametrize(
75 ("schema_id", "schema"),
76 [pytest.param(schema_id, schema, id=schema_id) for schema_id, schema in SCHEMAS],
77)
78def test_all_discovered_schemas_are_strict_portable(
79 schema_id: str, schema: dict[str, Any]
80) -> None:
81 schema_violations = violations(schema)
82 assert schema_violations == [], f"{schema_id}: {schema_violations}"
83
84
85@pytest.mark.parametrize("provider", STRICT_PROVIDERS)
86@pytest.mark.parametrize(
87 ("schema_id", "schema"),
88 [pytest.param(schema_id, schema, id=schema_id) for schema_id, schema in SCHEMAS],
89)
90def test_prepared_schemas_have_no_provider_unsupported_keywords(
91 schema_id: str, schema: dict[str, Any], provider: str
92) -> None:
93 prepared = prepare_provider_schema(schema, provider)
94
95 assert prepared is not None
96 assert unsupported_keyword_hits(prepared, provider) == [], schema_id
97
98
99@pytest.mark.parametrize(
100 "schema_path",
101 [
102 Path("solstone/talent/schedule.schema.json"),
103 Path("solstone/talent/sense.schema.json"),
104 ],
105)
106def test_zero_facet_runtime_hydration_of_shipped_schemas_has_no_banned_keys(
107 schema_path: Path, monkeypatch
108) -> None:
109 monkeypatch.setattr("solstone.think.talent._valid_runtime_facets", lambda: [])
110 schema = json.loads((REPO_ROOT / schema_path).read_text(encoding="utf-8"))
111
112 hydrated = hydrate_runtime_enums(schema)
113
114 for provider in STRICT_PROVIDERS:
115 prepared = prepare_provider_schema(hydrated, provider)
116 assert prepared is not None
117 assert unsupported_keyword_hits(prepared, provider) == []
118
119
120@pytest.mark.parametrize(
121 "schema",
122 [
123 {
124 "type": "object",
125 "properties": {
126 "a": {"type": "array"},
127 "b": {"type": "string"},
128 "c": {"oneOf": [{"type": "string"}, {"type": "integer"}]},
129 },
130 "required": ["a"],
131 }
132 ],
133)
134def test_strict_portability_guard_rejects_bad_schema(schema: dict[str, Any]) -> None:
135 schema_violations = violations(schema)
136
137 assert any(
138 "object missing additionalProperties:false" in v for v in schema_violations
139 )
140 assert any("properties not required" in v for v in schema_violations)
141 assert any("banned key 'oneOf'" in v for v in schema_violations)