1- from __future__ import annotations
2-
3- import argparse
4- import json
5- from dataclasses import dataclass
6- from datetime import UTC , datetime , timedelta
7- from pathlib import Path
8- from typing import Any
9-
10- from src .portfolio_truth_render import GENERATED_MARKDOWN_PROVENANCE_MARKER
11- from src .portfolio_truth_types import SCHEMA_VERSION , truth_latest_path
12-
13- DEFAULT_MAX_STALENESS_HOURS = 30
14- WORKLIST_SCHEMA_VERSION = "operator_os_seam_linter_worklist.v1"
15-
16- # v0.1: identity-resolution check - blocked on dialect census.
17- IDENTITY_RESOLUTION_EXTENSION_POINT = (
18- "v0.1: identity-resolution check - blocked on dialect census"
19- )
20-
21-
22- @dataclass (frozen = True )
23- class SeamLintFinding :
24- check : str
25- artifact : str
26- violation : str
27- detail : str
28-
29- def to_dict (self ) -> dict [str , str ]:
30- return {
31- "check" : self .check ,
32- "artifact" : self .artifact ,
33- "violation" : self .violation ,
34- "detail" : self .detail ,
35- }
36-
37-
38- @dataclass (frozen = True )
39- class SeamLintResult :
40- generated_at : datetime
41- expected_schema_version : str
42- max_staleness_hours : int
43- findings : tuple [SeamLintFinding , ...]
44-
45- @property
46- def passed (self ) -> bool :
47- return not self .findings
48-
49- def to_dict (self ) -> dict [str , Any ]:
50- return {
51- "schema_version" : "operator_os_seam_linter.v0" ,
52- "generated_at" : self .generated_at .isoformat (),
53- "state" : "pass" if self .passed else "fail" ,
54- "expected_schema_version" : self .expected_schema_version ,
55- "max_staleness_hours" : self .max_staleness_hours ,
56- "extension_points" : [IDENTITY_RESOLUTION_EXTENSION_POINT ],
57- "findings" : [finding .to_dict () for finding in self .findings ],
58- }
59-
60-
61- def lint_operator_os_seams (
62- * ,
63- truth_path : Path ,
64- markdown_paths : list [Path ],
65- expected_schema_version : str = SCHEMA_VERSION ,
66- max_staleness_hours : int = DEFAULT_MAX_STALENESS_HOURS ,
67- now : datetime | None = None ,
68- ) -> SeamLintResult :
69- generated_at = _aware (now or datetime .now (UTC ))
70- findings : list [SeamLintFinding ] = []
71- truth = _load_truth_artifact (truth_path , findings )
72- if truth is not None :
73- findings .extend (
74- _check_artifact_freshness (
75- truth ,
76- truth_path = truth_path ,
77- now = generated_at ,
78- max_staleness_hours = max_staleness_hours ,
79- )
80- )
81- findings .extend (
82- _check_schema_pin (
83- truth ,
84- truth_path = truth_path ,
85- expected_schema_version = expected_schema_version ,
86- )
87- )
88- findings .extend (_check_generated_markdown (markdown_paths ))
89- return SeamLintResult (
90- generated_at = generated_at ,
91- expected_schema_version = expected_schema_version ,
92- max_staleness_hours = max_staleness_hours ,
93- findings = tuple (findings ),
94- )
95-
96-
97- def build_worklist_payload (result : SeamLintResult ) -> dict [str , Any ]:
98- items = []
99- for finding in result .findings :
100- items .append (
101- {
102- "item_id" : f"ghra_seam_linter:{ finding .check } :{ Path (finding .artifact ).name } " ,
103- "kind" : "operator_os_seam_linter" ,
104- "severity" : "critical" ,
105- "title" : f"Operator-OS seam-linter failed: { finding .check } " ,
106- "summary" : f"{ finding .artifact } : { finding .violation } . { finding .detail } " ,
107- "target_type" : "artifact" ,
108- "target_id" : finding .artifact ,
109- "created_at" : result .generated_at .isoformat (),
110- "suggested_command" : (
111- "uv run python -m src.operator_os_seam_linter "
112- "--truth output/portfolio-truth-latest.json"
113- ),
114- "metadata_json" : json .dumps (finding .to_dict (), sort_keys = True ),
115- }
116- )
117- return {
118- "schema_version" : WORKLIST_SCHEMA_VERSION ,
119- "generated_at" : result .generated_at .isoformat (),
120- "state" : "pass" if result .passed else "fail" ,
121- "source" : "GithubRepoAuditor" ,
122- "items" : items ,
123- }
124-
125-
126- def main (argv : list [str ] | None = None ) -> int :
127- parser = _build_parser ()
128- args = parser .parse_args (argv )
129- workspace_root = Path (args .workspace_root ).expanduser ()
130- output_dir = Path (args .output_dir ).expanduser ()
131- truth_path = Path (args .truth ).expanduser () if args .truth else truth_latest_path (output_dir )
132- markdown_paths = [Path (path ).expanduser () for path in args .markdown ]
133- if not markdown_paths :
134- markdown_paths = [
135- workspace_root / "project-registry.md" ,
136- workspace_root / "PORTFOLIO-AUDIT-REPORT.md" ,
137- ]
138- result = lint_operator_os_seams (
139- truth_path = truth_path ,
140- markdown_paths = markdown_paths ,
141- expected_schema_version = args .expected_schema_version ,
142- max_staleness_hours = args .max_staleness_hours ,
143- )
144- payload = result .to_dict ()
145- if args .worklist_output :
146- worklist_path = Path (args .worklist_output ).expanduser ()
147- worklist_path .parent .mkdir (parents = True , exist_ok = True )
148- worklist_path .write_text (json .dumps (build_worklist_payload (result ), indent = 2 ) + "\n " )
149- if args .json :
150- print (json .dumps (payload , indent = 2 ))
151- else :
152- _print_text_summary (result )
153- return 0 if result .passed else 1
154-
155-
156- def _build_parser () -> argparse .ArgumentParser :
157- parser = argparse .ArgumentParser (
158- prog = "operator-os-seam-linter" ,
159- description = "Lint the small Operator-OS seam contract owned by GithubRepoAuditor." ,
160- )
161- parser .add_argument ("--output-dir" , type = Path , default = Path ("output" ))
162- parser .add_argument ("--workspace-root" , type = Path , default = Path .home () / "Projects" )
163- parser .add_argument ("--truth" , type = Path , default = None )
164- parser .add_argument ("--markdown" , action = "append" , default = [])
165- parser .add_argument ("--worklist-output" , type = Path , default = None )
166- parser .add_argument ("--expected-schema-version" , default = SCHEMA_VERSION )
167- parser .add_argument ("--max-staleness-hours" , type = int , default = DEFAULT_MAX_STALENESS_HOURS )
168- parser .add_argument ("--json" , action = "store_true" )
169- return parser
170-
171-
172- def _load_truth_artifact (
173- truth_path : Path , findings : list [SeamLintFinding ]
174- ) -> dict [str , Any ] | None :
175- try :
176- data = json .loads (truth_path .read_text ())
177- except FileNotFoundError :
178- findings .append (
179- SeamLintFinding (
180- check = "artifact_freshness" ,
181- artifact = str (truth_path ),
182- violation = "truth artifact is missing" ,
183- detail = "Regenerate portfolio truth before downstream consumers read it." ,
184- )
185- )
186- return None
187- except json .JSONDecodeError as exc :
188- findings .append (
189- SeamLintFinding (
190- check = "schema_pin" ,
191- artifact = str (truth_path ),
192- violation = "truth artifact is not valid JSON" ,
193- detail = str (exc ),
194- )
195- )
196- return None
197- if not isinstance (data , dict ):
198- findings .append (
199- SeamLintFinding (
200- check = "schema_pin" ,
201- artifact = str (truth_path ),
202- violation = "truth artifact root is not an object" ,
203- detail = "Expected the portfolio truth JSON object contract." ,
204- )
205- )
206- return None
207- return data
208-
209-
210- def _check_artifact_freshness (
211- truth : dict [str , Any ],
212- * ,
213- truth_path : Path ,
214- now : datetime ,
215- max_staleness_hours : int ,
216- ) -> list [SeamLintFinding ]:
217- raw_generated_at = truth .get ("generated_at" )
218- if not isinstance (raw_generated_at , str ) or not raw_generated_at .strip ():
219- return [
220- SeamLintFinding (
221- check = "artifact_freshness" ,
222- artifact = str (truth_path ),
223- violation = "generated_at is missing" ,
224- detail = "The truth artifact must declare when it was generated." ,
225- )
226- ]
227- try :
228- generated_at = _parse_datetime (raw_generated_at )
229- except ValueError as exc :
230- return [
231- SeamLintFinding (
232- check = "artifact_freshness" ,
233- artifact = str (truth_path ),
234- violation = "generated_at is not parseable" ,
235- detail = str (exc ),
236- )
237- ]
238- max_age = timedelta (hours = max_staleness_hours )
239- age = now - generated_at
240- if age > max_age :
241- return [
242- SeamLintFinding (
243- check = "artifact_freshness" ,
244- artifact = str (truth_path ),
245- violation = "truth artifact is stale" ,
246- detail = (
247- f"generated_at={ raw_generated_at } ; "
248- f"age_hours={ age .total_seconds () / 3600 :.2f} ; "
249- f"max_staleness_hours={ max_staleness_hours } "
250- ),
251- )
252- ]
253- if generated_at - now > timedelta (minutes = 5 ):
254- return [
255- SeamLintFinding (
256- check = "artifact_freshness" ,
257- artifact = str (truth_path ),
258- violation = "generated_at is in the future" ,
259- detail = f"generated_at={ raw_generated_at } ; now={ now .isoformat ()} " ,
260- )
261- ]
262- return []
263-
264-
265- def _check_schema_pin (
266- truth : dict [str , Any ],
267- * ,
268- truth_path : Path ,
269- expected_schema_version : str ,
270- ) -> list [SeamLintFinding ]:
271- declared = truth .get ("schema_version" )
272- if not isinstance (declared , str ) or not declared .strip ():
273- return [
274- SeamLintFinding (
275- check = "schema_pin" ,
276- artifact = str (truth_path ),
277- violation = "schema_version is missing" ,
278- detail = "The truth artifact must declare its schema pin." ,
279- )
280- ]
281- if declared != expected_schema_version :
282- return [
283- SeamLintFinding (
284- check = "schema_pin" ,
285- artifact = str (truth_path ),
286- violation = "schema_version mismatch" ,
287- detail = f"declared={ declared } ; expected={ expected_schema_version } " ,
288- )
289- ]
290- return []
291-
292-
293- def _check_generated_markdown (markdown_paths : list [Path ]) -> list [SeamLintFinding ]:
294- findings : list [SeamLintFinding ] = []
295- for path in markdown_paths :
296- try :
297- text = path .read_text ()
298- except FileNotFoundError :
299- findings .append (
300- SeamLintFinding (
301- check = "markdown_generated_not_hand_edited" ,
302- artifact = str (path ),
303- violation = "generated markdown is missing" ,
304- detail = "Expected a generated compatibility markdown artifact." ,
305- )
306- )
307- continue
308- if GENERATED_MARKDOWN_PROVENANCE_MARKER not in text :
309- findings .append (
310- SeamLintFinding (
311- check = "markdown_generated_not_hand_edited" ,
312- artifact = str (path ),
313- violation = "generated provenance marker is missing" ,
314- detail = (
315- "Regenerate this markdown with GithubRepoAuditor; "
316- "the linter does not guess via content diffs."
317- ),
318- )
319- )
320- return findings
321-
322-
323- def _parse_datetime (value : str ) -> datetime :
324- normalized = value .strip ()
325- if normalized .endswith ("Z" ):
326- normalized = f"{ normalized [:- 1 ]} +00:00"
327- parsed = datetime .fromisoformat (normalized )
328- return _aware (parsed )
329-
330-
331- def _aware (value : datetime ) -> datetime :
332- if value .tzinfo is None :
333- return value .replace (tzinfo = UTC )
334- return value .astimezone (UTC )
335-
336-
337- def _print_text_summary (result : SeamLintResult ) -> None :
338- state = "PASS" if result .passed else "FAIL"
339- print (f"Operator-OS seam-linter: { state } " )
340- print (f"expected_schema_version={ result .expected_schema_version } " )
341- print (f"max_staleness_hours={ result .max_staleness_hours } " )
342- if not result .findings :
343- print ("No seam violations found." )
344- return
345- for finding in result .findings :
346- print (f"- { finding .check } : { finding .artifact } : { finding .violation } " )
347- print (f" { finding .detail } " )
348-
349-
350- if __name__ == "__main__" :
351- raise SystemExit (main ())
1+ @src / operator_os_seam_linter .py
0 commit comments