feat(fleet): fiscal profile + compliance calendar + ADC register (erp#54, T11 data)
fleet/profile/ goes from stub to the machine-readable business-rules surface the fleet reads (PRD agent-catalog document surface + compliance ADC framework): - fiscal.yaml — entity, VAT position, 8 rules (regime reel simplifie until 2026-12-31 -> quarterly CA3 from 2027-01-01 per LF 2025 art. 38; KM export autoliquidation 259-1 CGI box E2; FR 20% deductible; intra-EU reverse charge; FX 766/666; SaaS expensed; CCA 455 lane). Every rule carries effective_from/effective_until AND decision: adc-NNN; every date cites its PRD anchor as an inline comment (verified against factory origin/main). - calendar.yaml — 15 entries: acomptes TVA (2026-07 month-window, 2026-12-15), last CA12 FY-2026 (2027-05-04), CA3 quarterly windows, CFE (December), AG comptes annuels (2027-06-30), e-invoicing milestones (2026-09-01 reception, 2027-09-01 emission/e-reporting), URSSAF echeancier with the in-file NOTE that a real direct debit exists since May 2026 (erp#57 revisit of the payroll-dormant assumption), KM deferred due dates + renewal stub. - JSON Schemas for both + scripts/validate.py (stdlib-only: strict YAML-subset parser, JSON-Schema-subset checker, rule->ADC resolution, calendar checks). - decisions/ — ADC register: template + adc-001..005 Accepted formalizations (autoliquidation KM, FX->766/666, SaaS expensed, reel simplifie until abolition, CCA personal-card lane) + adc-006/007 Proposed stubs (retainer currency -> erp#53; capital path -> erp#51). Agents draft, the operator Accepts — never the reverse; immutable once merged, supersede never edit. - Mutation policy in-file: PRs only (T12 proposes, human merges). - Same-change: profile README stub -> real doc; fleet/README.md layout line and AGENTS.md fleet row updated (profile no longer a stub). Validation: PASS — 8 rules, 15 entries, 7 ADCs, 0 errors, 7 warnings (the warnings list exactly what awaits operator verification). Human gate left open on purpose: operator sanity-read of the calendar + Acceptance of adc-001..005. Co-Authored-By: Claude Fable 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01VRShc4QhLLU73FLHx9vskh
This commit is contained in:
Claude Fable 5
@@ -0,0 +1,435 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Validate the fleet/profile document surface (erp#54).
|
||||
|
||||
Checks, in order:
|
||||
1. fiscal.yaml parses (strict YAML subset) and validates against fiscal.schema.json
|
||||
2. calendar.yaml parses and validates against calendar.schema.json
|
||||
3. Referential integrity:
|
||||
- every fiscal rule's `decision: adc-NNN` resolves to exactly one
|
||||
decisions/adc-NNN-*.md whose frontmatter status is Accepted
|
||||
(agents draft, the operator Accepts - a rule may only rely on an
|
||||
Accepted decision)
|
||||
- rule/entry ids unique; effective_from <= effective_until
|
||||
- calendar entries: exactly one of due|recurrence, unless status is
|
||||
pending-definition/conditional; recurrence months are 1..12
|
||||
4. ADC register hygiene: filename <-> frontmatter id match, valid status
|
||||
(Proposed | Accepted | Superseded-by-NNN), Accepted records carry
|
||||
decided + effective_from, Proposed records carry decided: null,
|
||||
required body sections present, no duplicate adc numbers.
|
||||
|
||||
Stdlib only, by design (PRD agent-catalog "document surface" + erp#54): the
|
||||
profile YAML is written in a strict subset - block maps, block sequences,
|
||||
single-line scalars, one-line [flow] lists of scalars, comments; no anchors,
|
||||
no multi-line scalars, no nested flow - and the schemas use the JSON-Schema
|
||||
subset {type, properties, required, additionalProperties, items, enum,
|
||||
pattern}.
|
||||
|
||||
Usage: python3 fleet/profile/scripts/validate.py
|
||||
Exit 0 = all green (warnings allowed). Exit 1 = at least one error.
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
PROFILE = Path(__file__).resolve().parent.parent
|
||||
DECISIONS = PROFILE / "decisions"
|
||||
|
||||
DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
|
||||
ADC_FILE_RE = re.compile(r"^adc-(\d{3})-[a-z0-9-]+\.md$")
|
||||
ADC_STATUS_RE = re.compile(r"^(Proposed|Accepted|Superseded-by-\d{3})$")
|
||||
ADC_REQUIRED_SECTIONS = [
|
||||
"## Context",
|
||||
"## Decision",
|
||||
"## Base légale & doctrine",
|
||||
"## Alternatives rejected",
|
||||
"## Consequences",
|
||||
"## QA & validation",
|
||||
"## References",
|
||||
]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Strict YAML-subset parser
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class YamlSubsetError(Exception):
|
||||
def __init__(self, msg, line=None):
|
||||
super().__init__(f"line {line}: {msg}" if line else msg)
|
||||
|
||||
|
||||
def _strip_comment(raw, line):
|
||||
"""Return the value token of a raw value string, comment stripped."""
|
||||
s = raw.strip()
|
||||
if not s or s.startswith("#"):
|
||||
return ""
|
||||
if s[0] in "\"'":
|
||||
q = s[0]
|
||||
end = s.find(q, 1)
|
||||
if end == -1:
|
||||
raise YamlSubsetError("unterminated quoted scalar", line)
|
||||
rest = s[end + 1:].strip()
|
||||
if rest and not rest.startswith("#"):
|
||||
raise YamlSubsetError(f"trailing content after quoted scalar: {rest!r}", line)
|
||||
return s[: end + 1]
|
||||
idx = s.find(" #")
|
||||
if idx != -1:
|
||||
s = s[:idx]
|
||||
return s.strip()
|
||||
|
||||
|
||||
def _scalar(tok, line):
|
||||
if tok == "" or tok in ("null", "~"):
|
||||
return None
|
||||
if tok == "true":
|
||||
return True
|
||||
if tok == "false":
|
||||
return False
|
||||
if len(tok) >= 2 and tok[0] in "\"'" and tok[-1] == tok[0]:
|
||||
return tok[1:-1]
|
||||
if tok.startswith("["):
|
||||
if not tok.endswith("]"):
|
||||
raise YamlSubsetError("unterminated flow list", line)
|
||||
inner = tok[1:-1].strip()
|
||||
if not inner:
|
||||
return []
|
||||
return [_scalar(p.strip(), line) for p in inner.split(",")]
|
||||
if re.fullmatch(r"-?\d+", tok):
|
||||
return int(tok)
|
||||
if re.fullmatch(r"-?\d+\.\d+", tok):
|
||||
return float(tok)
|
||||
return tok # plain string (ISO dates stay strings)
|
||||
|
||||
|
||||
def parse_yaml(text, name="<yaml>"):
|
||||
items = []
|
||||
for n, raw in enumerate(text.splitlines(), 1):
|
||||
if not raw.strip() or raw.strip().startswith("#"):
|
||||
continue
|
||||
if "\t" in raw:
|
||||
raise YamlSubsetError(f"{name}: tab character (use spaces)", n)
|
||||
indent = len(raw) - len(raw.lstrip(" "))
|
||||
items.append((indent, raw.strip(), n))
|
||||
if not items:
|
||||
return {}
|
||||
value, nxt = _parse_block(items, 0, items[0][0])
|
||||
if nxt != len(items):
|
||||
raise YamlSubsetError(f"{name}: trailing content", items[nxt][2])
|
||||
return value
|
||||
|
||||
|
||||
def _parse_block(items, i, indent):
|
||||
_, content, line = items[i]
|
||||
if items[i][0] != indent:
|
||||
raise YamlSubsetError("unexpected indent", line)
|
||||
if content == "-" or content.startswith("- "):
|
||||
return _parse_seq(items, i, indent)
|
||||
return _parse_map(items, i, indent)
|
||||
|
||||
|
||||
def _parse_map(items, i, indent):
|
||||
result = {}
|
||||
while i < len(items):
|
||||
ind, content, line = items[i]
|
||||
if ind < indent:
|
||||
break
|
||||
if ind > indent:
|
||||
raise YamlSubsetError("unexpected deeper indent", line)
|
||||
if content == "-" or content.startswith("- "):
|
||||
raise YamlSubsetError("sequence item at mapping level", line)
|
||||
m = re.match(r"^([A-Za-z_][A-Za-z0-9_-]*):(?:\s+(.*))?$", content)
|
||||
if not m:
|
||||
raise YamlSubsetError(f"not a 'key: value' line: {content!r}", line)
|
||||
key, rest = m.group(1), m.group(2)
|
||||
if key in result:
|
||||
raise YamlSubsetError(f"duplicate key {key!r}", line)
|
||||
tok = _strip_comment(rest, line) if rest else ""
|
||||
if tok:
|
||||
result[key] = _scalar(tok, line)
|
||||
i += 1
|
||||
else:
|
||||
i += 1
|
||||
if i < len(items) and items[i][0] > indent:
|
||||
value, i = _parse_block(items, i, items[i][0])
|
||||
result[key] = value
|
||||
else:
|
||||
result[key] = None
|
||||
return result, i
|
||||
|
||||
|
||||
def _parse_seq(items, i, indent):
|
||||
result = []
|
||||
while i < len(items):
|
||||
ind, content, line = items[i]
|
||||
if ind < indent:
|
||||
break
|
||||
if ind > indent:
|
||||
raise YamlSubsetError("unexpected deeper indent in sequence", line)
|
||||
if not (content == "-" or content.startswith("- ")):
|
||||
break
|
||||
rest = content[1:].strip()
|
||||
if not rest or rest.startswith("#"):
|
||||
i += 1
|
||||
if i < len(items) and items[i][0] > indent:
|
||||
value, i = _parse_block(items, i, items[i][0])
|
||||
result.append(value)
|
||||
else:
|
||||
result.append(None)
|
||||
elif re.match(r"^[A-Za-z_][A-Za-z0-9_-]*:(\s|$)", rest):
|
||||
# inline first key of a mapping item: reparse at indent+2
|
||||
sub = [(indent + 2, rest, line)]
|
||||
i += 1
|
||||
while i < len(items) and items[i][0] > indent:
|
||||
s_ind, s_content, s_line = items[i]
|
||||
if s_ind < indent + 2:
|
||||
raise YamlSubsetError("bad indent inside sequence item", s_line)
|
||||
sub.append((s_ind, s_content, s_line))
|
||||
i += 1
|
||||
value, used = _parse_map(sub, 0, indent + 2)
|
||||
if used != len(sub):
|
||||
raise YamlSubsetError("trailing content in sequence item", sub[used][2])
|
||||
result.append(value)
|
||||
else:
|
||||
tok = _strip_comment(rest, line)
|
||||
result.append(_scalar(tok, line))
|
||||
i += 1
|
||||
return result, i
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# JSON-Schema subset checker
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def _is_type(v, t):
|
||||
return {
|
||||
"object": lambda: isinstance(v, dict),
|
||||
"array": lambda: isinstance(v, list),
|
||||
"string": lambda: isinstance(v, str),
|
||||
"integer": lambda: isinstance(v, int) and not isinstance(v, bool),
|
||||
"number": lambda: isinstance(v, (int, float)) and not isinstance(v, bool),
|
||||
"boolean": lambda: isinstance(v, bool),
|
||||
"null": lambda: v is None,
|
||||
}[t]()
|
||||
|
||||
|
||||
def schema_check(value, schema, path, errors):
|
||||
types = schema.get("type")
|
||||
if types:
|
||||
if isinstance(types, str):
|
||||
types = [types]
|
||||
if not any(_is_type(value, t) for t in types):
|
||||
errors.append(f"{path}: expected {'/'.join(types)}, got {type(value).__name__}")
|
||||
return
|
||||
if value is None:
|
||||
return
|
||||
if "enum" in schema and value not in schema["enum"]:
|
||||
errors.append(f"{path}: {value!r} not in {schema['enum']}")
|
||||
if isinstance(value, str) and "pattern" in schema:
|
||||
if not re.search(schema["pattern"], value):
|
||||
errors.append(f"{path}: {value!r} does not match {schema['pattern']!r}")
|
||||
if isinstance(value, dict):
|
||||
props = schema.get("properties", {})
|
||||
for req in schema.get("required", []):
|
||||
if req not in value:
|
||||
errors.append(f"{path}: missing required key '{req}'")
|
||||
if schema.get("additionalProperties") is False:
|
||||
for k in value:
|
||||
if k not in props:
|
||||
errors.append(f"{path}: unexpected key '{k}'")
|
||||
for k, v in value.items():
|
||||
if k in props:
|
||||
schema_check(v, props[k], f"{path}.{k}", errors)
|
||||
if isinstance(value, list) and "items" in schema:
|
||||
for idx, item in enumerate(value):
|
||||
schema_check(item, schema["items"], f"{path}[{idx}]", errors)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# ADC register
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def parse_frontmatter(path):
|
||||
text = path.read_text(encoding="utf-8")
|
||||
if not text.startswith("---\n"):
|
||||
raise YamlSubsetError(f"{path.name}: missing frontmatter")
|
||||
end = text.find("\n---", 4)
|
||||
if end == -1:
|
||||
raise YamlSubsetError(f"{path.name}: unterminated frontmatter")
|
||||
return parse_yaml(text[4:end], path.name), text[end + 4:]
|
||||
|
||||
|
||||
def load_adc_register(errors):
|
||||
"""Return {'adc-NNN': {'file': name, 'status': str, ...}}."""
|
||||
register = {}
|
||||
if not DECISIONS.is_dir():
|
||||
errors.append(f"decisions/ directory missing at {DECISIONS}")
|
||||
return register
|
||||
for path in sorted(DECISIONS.glob("*.md")):
|
||||
if path.name == "adc-template.md":
|
||||
continue
|
||||
m = ADC_FILE_RE.match(path.name)
|
||||
if not m:
|
||||
errors.append(f"decisions/{path.name}: name must match adc-NNN-<slug>.md")
|
||||
continue
|
||||
adc_id = f"adc-{m.group(1)}"
|
||||
if adc_id in register:
|
||||
errors.append(f"decisions/{path.name}: duplicate id {adc_id} "
|
||||
f"(also {register[adc_id]['file']})")
|
||||
continue
|
||||
try:
|
||||
fm, body = parse_frontmatter(path)
|
||||
except YamlSubsetError as e:
|
||||
errors.append(f"decisions/{path.name}: {e}")
|
||||
continue
|
||||
if fm.get("id") != adc_id:
|
||||
errors.append(f"decisions/{path.name}: frontmatter id {fm.get('id')!r} != {adc_id}")
|
||||
status = fm.get("status")
|
||||
if not (isinstance(status, str) and ADC_STATUS_RE.match(status)):
|
||||
errors.append(f"decisions/{path.name}: invalid status {status!r} "
|
||||
"(Proposed | Accepted | Superseded-by-NNN)")
|
||||
if status == "Accepted":
|
||||
if not (isinstance(fm.get("decided"), str) and DATE_RE.match(fm["decided"])):
|
||||
errors.append(f"decisions/{path.name}: Accepted requires a 'decided' date")
|
||||
if not (isinstance(fm.get("effective_from"), str) and DATE_RE.match(fm["effective_from"])):
|
||||
errors.append(f"decisions/{path.name}: Accepted requires 'effective_from'")
|
||||
elif status == "Proposed":
|
||||
if fm.get("decided") is not None:
|
||||
errors.append(f"decisions/{path.name}: Proposed must keep decided: null "
|
||||
"(acceptance is a human act)")
|
||||
for section in ADC_REQUIRED_SECTIONS:
|
||||
if section not in body:
|
||||
errors.append(f"decisions/{path.name}: missing section '{section}'")
|
||||
register[adc_id] = dict(fm, file=path.name)
|
||||
return register
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Custom (referential) checks
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def check_dates_ordered(obj, path, errors):
|
||||
frm, until = obj.get("effective_from"), obj.get("effective_until")
|
||||
if isinstance(frm, str) and isinstance(until, str) and frm > until:
|
||||
errors.append(f"{path}: effective_from {frm} > effective_until {until}")
|
||||
|
||||
|
||||
def check_fiscal(fiscal, register, errors, warnings, resolution):
|
||||
seen = set()
|
||||
for idx, rule in enumerate(fiscal.get("rules") or []):
|
||||
rid = rule.get("id", f"[{idx}]")
|
||||
path = f"fiscal.rules.{rid}"
|
||||
if rid in seen:
|
||||
errors.append(f"{path}: duplicate rule id")
|
||||
seen.add(rid)
|
||||
check_dates_ordered(rule, path, errors)
|
||||
decision = rule.get("decision")
|
||||
adc = register.get(decision)
|
||||
if adc is None:
|
||||
errors.append(f"{path}: decision {decision!r} does not resolve to any "
|
||||
f"decisions/adc-NNN-*.md record")
|
||||
resolution.append((rid, decision, "<missing>", "<missing>"))
|
||||
continue
|
||||
resolution.append((rid, decision, adc["file"], adc.get("status")))
|
||||
if adc.get("status") != "Accepted":
|
||||
errors.append(f"{path}: decision {decision} has status "
|
||||
f"{adc.get('status')!r} - a rule in force may only cite an "
|
||||
"Accepted decision (agents draft, the operator Accepts)")
|
||||
|
||||
|
||||
def check_calendar(calendar, errors, warnings):
|
||||
seen = set()
|
||||
for idx, entry in enumerate(calendar.get("entries") or []):
|
||||
eid = entry.get("id", f"[{idx}]")
|
||||
path = f"calendar.entries.{eid}"
|
||||
if eid in seen:
|
||||
errors.append(f"{path}: duplicate entry id")
|
||||
seen.add(eid)
|
||||
check_dates_ordered(entry, path, errors)
|
||||
status = entry.get("status", "confirmed")
|
||||
has_due = entry.get("due") is not None
|
||||
has_rec = entry.get("recurrence") is not None
|
||||
if has_due and has_rec:
|
||||
errors.append(f"{path}: carries both due and recurrence")
|
||||
elif not has_due and not has_rec and status not in ("pending-definition", "conditional"):
|
||||
errors.append(f"{path}: needs due or recurrence (status {status!r})")
|
||||
rec = entry.get("recurrence")
|
||||
months = entry.get("months")
|
||||
if rec in ("yearly", "quarterly"):
|
||||
if not months:
|
||||
errors.append(f"{path}: recurrence {rec!r} requires months")
|
||||
elif months is not None:
|
||||
errors.append(f"{path}: months only allowed with yearly/quarterly recurrence")
|
||||
for mth in months or []:
|
||||
if not (isinstance(mth, int) and 1 <= mth <= 12):
|
||||
errors.append(f"{path}: month {mth!r} not in 1..12")
|
||||
if has_due and re.fullmatch(r"\d{4}-\d{2}", entry["due"]):
|
||||
warnings.append(f"{path}: month-precision due ({entry['due']}) - the source "
|
||||
"gives no day; verify on the authority's notice")
|
||||
if status in ("estimated", "conditional", "pending-definition"):
|
||||
warnings.append(f"{path}: status {status} - operator verification pending")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Main
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def load_and_validate(name, errors):
|
||||
yaml_path = PROFILE / f"{name}.yaml"
|
||||
schema_path = PROFILE / f"{name}.schema.json"
|
||||
try:
|
||||
data = parse_yaml(yaml_path.read_text(encoding="utf-8"), yaml_path.name)
|
||||
except (OSError, YamlSubsetError) as e:
|
||||
errors.append(f"{yaml_path.name}: {e}")
|
||||
return None
|
||||
try:
|
||||
schema = json.loads(schema_path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
errors.append(f"{schema_path.name}: {e}")
|
||||
return None
|
||||
before = len(errors)
|
||||
schema_check(data, schema, name, errors)
|
||||
print(f"== {yaml_path.name} ==")
|
||||
print(f" parse: OK")
|
||||
print(f" schema ({schema_path.name}): {'OK' if len(errors) == before else 'FAIL'}")
|
||||
return data
|
||||
|
||||
|
||||
def main():
|
||||
errors, warnings, resolution = [], [], []
|
||||
|
||||
fiscal = load_and_validate("fiscal", errors)
|
||||
calendar = load_and_validate("calendar", errors)
|
||||
|
||||
print("== decisions/ (ADC register) ==")
|
||||
register = load_adc_register(errors)
|
||||
for adc_id in sorted(register):
|
||||
adc = register[adc_id]
|
||||
print(f" {adc['file']:<44} [{adc.get('status')}]")
|
||||
|
||||
if fiscal is not None:
|
||||
check_fiscal(fiscal, register, errors, warnings, resolution)
|
||||
print("== rule -> ADC resolution ==")
|
||||
for rid, decision, fname, status in resolution:
|
||||
print(f" {rid:<32} -> {decision} -> {fname} [{status}]")
|
||||
if calendar is not None:
|
||||
check_calendar(calendar, errors, warnings)
|
||||
|
||||
if warnings:
|
||||
print(f"== warnings ({len(warnings)}) ==")
|
||||
for w in warnings:
|
||||
print(f" WARN {w}")
|
||||
if errors:
|
||||
print(f"== errors ({len(errors)}) ==")
|
||||
for e in errors:
|
||||
print(f" ERROR {e}")
|
||||
n_rules = len((fiscal or {}).get("rules") or [])
|
||||
n_entries = len((calendar or {}).get("entries") or [])
|
||||
verdict = "FAIL" if errors else "PASS"
|
||||
print(f"== RESULT: {verdict} - {n_rules} rules, {n_entries} calendar entries, "
|
||||
f"{len(register)} ADC records, {len(errors)} errors, {len(warnings)} warnings ==")
|
||||
return 1 if errors else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user