Define reviewed publications and record the field-sampling frame.

IR-WP-0007-T03 can now record a capture-linked publication separately from generated drafts. The field-outcome design is written, and credential, billing, and website-pull gates were rechecked and left waiting.

Assistant: grok
Assistant-Session: 01a106a5-fc07-78c2-b64f-8e04cea25daf

intelligence_radar/__main__.py

@@ -154,6 +154,8 @@ def check(root):
         all_records(root, kind)
     from .external_briefs import verify
     verify(root)
+    from .publications import verify as verify_publications
+    verify_publications(root)
     from .advisor import replay
     for ident, _ in all_records(root, 'recommendations'):
         replay(root, ident)
@@ -189,6 +191,8 @@ def main(argv=None):
     briefs_command = commands.add_parser('import-briefs', help='retain unarchived ChatGPT and Grok briefs from the local inbox')
     briefs_command.add_argument('--config', type=Path, default=Path('ops/schedule.json'))
     briefs_command.add_argument('--inbox', type=Path, help='override the configured inbox for this run')
+    publish_command = commands.add_parser('publish', help='record a reviewed publication linked to captures')
+    publish_command.add_argument('input', type=Path)
     commands.add_parser('health', help='show durable scheduler health')
     export_command = commands.add_parser('export', help='export a recommendation and source dependencies as a tar bundle')
     export_command.add_argument('id')
@@ -249,6 +253,9 @@ def main(argv=None):
             print(json.dumps(result, indent=2))
             external = result.get('external_briefs') or {}
             return 0 if result['ok'] and external.get('ok', True) else 1
+        elif args.command == 'publish':
+            from .publications import publish
+            print(publish(args.root, json.loads(args.input.read_text())))
         elif args.command == 'import-briefs':
             from .external_briefs import import_inbox, native_status, resolve_inbox
             spec = json.loads(args.config.read_text()).get('external_briefs', {})

intelligence_radar/publications.py

@@ -0,0 +1,113 @@
+"""Reviewed publications. A generated digest is not one of these records."""
+from datetime import datetime
+from pathlib import Path
+
+from .store import all_records, get, put
+
+FIELDS = {'schema_version', 'title', 'reviewer', 'reviewed_at', 'claims', 'supersedes'}
+CLAIM_FIELDS = {'statement', 'disposition', 'capture_id', 'exclusion_reason'}
+
+
+def publish(root, value):
+    root = Path(root)
+    record = _record(root, value)
+    ident = put(root, 'publications', record)
+    _render(root, ident, record)
+    return ident
+
+
+def verify(root):
+    root = Path(root)
+    directory = root / 'data' / 'publications'
+    if not directory.exists():
+        return 0
+    reviewed = (root / 'digests' / 'reviewed').resolve()
+    count = 0
+    for ident, record in all_records(root, 'publications'):
+        _record(root, record, existing=ident)
+        path = (reviewed / f'{ident}.md').resolve()
+        if path.parent != reviewed or not path.is_file():
+            raise ValueError(f'reviewed publication file missing: {ident}')
+        text = path.read_text(encoding='utf-8')
+        if not text.startswith('# Reviewed publication\n') or ident not in text:
+            raise ValueError(f'reviewed publication rendering mismatch: {ident}')
+        count += 1
+    return count
+
+
+def _record(root, value, existing=None):
+    if not isinstance(value, dict) or set(value) - FIELDS or FIELDS - {'supersedes'} - set(value):
+        raise ValueError('publication has missing or unknown fields')
+    if type(value['schema_version']) is not int or value['schema_version'] != 1:
+        raise ValueError('schema_version must be 1')
+    for key in ('title', 'reviewer'):
+        if not isinstance(value[key], str) or not value[key].strip():
+            raise ValueError(f'{key} must be nonempty text')
+    reviewed = datetime.fromisoformat(value['reviewed_at'].replace('Z', '+00:00'))
+    if reviewed.tzinfo is None:
+        raise ValueError('reviewed_at requires a timezone')
+    if not isinstance(value['claims'], list) or not value['claims']:
+        raise ValueError('publication requires at least one claim')
+    record = {'schema_version': 1, 'title': value['title'].strip(),
+              'reviewer': value['reviewer'].strip(), 'reviewed_at': value['reviewed_at'],
+              'claims': [_claim(root, claim) for claim in value['claims']]}
+    if 'supersedes' in value:
+        parent = value['supersedes']
+        if not isinstance(parent, str) or parent == existing:
+            raise ValueError('supersedes must name an earlier publication')
+        _chain(root, parent, set())
+        record['supersedes'] = parent
+    return record
+
+
+def _claim(root, claim):
+    if not isinstance(claim, dict) or set(claim) - CLAIM_FIELDS or {'statement', 'disposition'} - set(claim):
+        raise ValueError('invalid publication claim')
+    if not isinstance(claim['statement'], str) or not claim['statement'].strip():
+        raise ValueError('claim statement must be nonempty text')
+    if claim['disposition'] not in {'included', 'excluded'}:
+        raise ValueError('claim disposition must be included or excluded')
+    item = {'statement': claim['statement'].strip(), 'disposition': claim['disposition']}
+    if 'capture_id' in claim:
+        get(root, 'captures', claim['capture_id'])
+        item['capture_id'] = claim['capture_id']
+    if claim['disposition'] == 'included':
+        if 'capture_id' not in item or 'exclusion_reason' in claim:
+            raise ValueError('an included claim requires a capture and no exclusion reason')
+    else:
+        reason = claim.get('exclusion_reason')
+        if not isinstance(reason, str) or not reason.strip():
+            raise ValueError('an excluded claim requires an exclusion reason')
+        item['exclusion_reason'] = reason.strip()
+    return item
+
+
+def _chain(root, ident, seen):
+    if ident in seen:
+        raise ValueError('publication correction cycle')
+    seen.add(ident)
+    try:
+        record = get(root, 'publications', ident)
+    except OSError as exc:
+        raise ValueError('supersedes must name an earlier publication') from exc
+    if 'supersedes' in record:
+        _chain(root, record['supersedes'], seen)
+
+
+def _render(root, ident, record):
+    lines = ['# Reviewed publication', '',
+             'This file is a reviewed publication. It is not a generated daily, weekly, or external draft.',
+             '', f'- Revision: `{ident}`', f"- Title: {record['title']}",
+             f"- Reviewer: {record['reviewer']}", f"- Reviewed: {record['reviewed_at']}"]
+    if 'supersedes' in record:
+        lines.append(f"- Supersedes: `{record['supersedes']}`")
+    lines += ['', '## Claims', '']
+    for claim in record['claims']:
+        lines.append(f"- {claim['disposition']}: {claim['statement']}")
+        if 'capture_id' in claim:
+            lines.append(f"  - Capture: `{claim['capture_id']}`")
+        if 'exclusion_reason' in claim:
+            lines.append(f"  - Excluded because: {claim['exclusion_reason']}")
+    path = Path(root) / 'digests' / 'reviewed' / f'{ident}.md'
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text('\n'.join(lines) + '\n', encoding='utf-8')