coulomb/intelligence-radar
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Retain briefs routed by inbound-library as partial mailed teasers.

Browse source 
IR-WP-0010-T03: pull-briefs and a run-due step read routed items through
the library CLI, keep the original message as the artifact, classify
cadence and period, and record schema-2 briefs marked partial with the
unfetched Continue-reading link. Retains the two 4 October Grok mails.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 7780@bnt-lap001
Assistant-Session: 622cd705-cb31-4322-ba08-dd42e44fa583
This commit is contained in:
tegwick 2026-10-06 00:28:11 +02:00
parent af86299186
commit 1d4c55ef25
13 changed files with 1398 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

14
intelligence_radar/__main__.py
View file

@ -119,7 +119,8 @@ def draft(root, period):
'Unverified ChatGPT or Grok automation output retained for this period.',
'These briefs are not reviewed facts.', '']
for brief in briefs:
lines += [f"- {brief['provider']} {brief['cadence']}: {brief['digest_path']}",
kind = ' (mailed teaser, partial)' if brief.get('completeness') == 'partial' else ''
lines += [f"- {brief['provider']} {brief['cadence']}{kind}: {brief['digest_path']}",
f"- Content: {brief['content_sha256']}"]
lines.append('')
lines += ['## Editorial follow-up', '',
@ -191,6 +192,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')
pull_command = commands.add_parser('pull-briefs', help='retain briefs that inbound-library routed to this repo')
pull_command.add_argument('--config', type=Path, default=Path('ops/schedule.json'))
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')
@ -265,6 +268,15 @@ def main(argv=None):
result['native'] = native_status()
print(json.dumps(result, indent=2))
return 0 if result['ok'] else 1
elif args.command == 'pull-briefs':
from .library_briefs import pull
spec = json.loads(args.config.read_text()).get('library')
if not spec:
print('no "library" section in the schedule config', file=sys.stderr)
return 2
result = pull(args.root, spec)
print(json.dumps(result, indent=2))
return 0 if result['ok'] else 1
elif args.command == 'health':
path = args.root / '.runtime/scheduler.json'
print(path.read_text() if path.exists() else '{"status": "not_run"}')

6
intelligence_radar/external_briefs.py
View file

@ -19,6 +19,10 @@ NAME = re.compile(
r'^(?P<ymd>\d{6})-intelligenceRadar(?P<cadence>Daily|Weekly)_(?P<provider>chatgpt|grok)\.txt$')
RECORD_FIELDS = {'schema_version', 'provider', 'cadence', 'period', 'filename_date',
'source_name', 'content_sha256', 'digest_path', 'imported_at'}
# Schema 2: a brief retained from an inbound-library mail (library_briefs.py).
RECORD_FIELDS_V2 = RECORD_FIELDS | {'origin', 'library_source', 'received_at', 'period_basis',
'completeness', 'continue_url'}
SCHEMAS = {1: RECORD_FIELDS, 2: RECORD_FIELDS_V2}
NATIVE_STATUS = (
{'provider': 'chatgpt', 'status': 'needs_session',
@ -81,7 +85,7 @@ def verify(root):
external = (root / 'digests' / 'external').resolve()
count = 0
for ident, record in all_records(root, 'external-briefs'):
if set(record) != RECORD_FIELDS or record['schema_version'] != 1:
if set(record) != SCHEMAS.get(record.get('schema_version')):
raise ValueError(f'invalid external brief: {ident}')
digest = record['content_sha256']
blob = (root / 'data' / 'artifacts' / digest).read_bytes()

168
intelligence_radar/library_briefs.py Normal file
View file

@ -0,0 +1,168 @@
"""Retain automation briefs that inbound-library routed to this repository.
The library is reached only through its pull CLI (`new`, `show`, `get`). Grok
mails a notification rather than the brief: the automation name, the title, a
teaser cut off mid-sentence, and a "Continue reading" link that needs a
signed-in session. Such a mail is retained as a `partial` brief, linked to its
period. The full text still arrives through the drop folder (external_briefs).
No session is sent to grok.com.
"""
import hashlib
import json
import re
import subprocess
import sys
from datetime import date, datetime
from pathlib import Path
from zoneinfo import ZoneInfo
from .ingest import MAX_BYTES
from .store import all_records, atomic, put
AUTOMATION = re.compile(r'intelligence-radar-(daily|weekly)', re.IGNORECASE)
CONTINUE = re.compile(r'Continue reading\s*<?(https://grok\.com/\S+?)>?\s', re.IGNORECASE)
FORWARD_MARKER = re.compile(r'-{3,}\s*(?:Forwarded message|Weitergeleitete Nachricht)\s*-{3,}',
re.IGNORECASE)
FORWARD_DATE = re.compile(r'^\s*(?:Date|Datum)\s*:\s*(.+)$', re.MULTILINE)
MONTHS = {'jan': 1, 'feb': 2, 'mar': 3, 'mär': 3, 'maer': 3, 'apr': 4, 'may': 5, 'mai': 5,
'jun': 6, 'jul': 7, 'aug': 8, 'sep': 9, 'oct': 10, 'okt': 10, 'nov': 11,
'dec': 12, 'dez': 12}
class LibraryUnavailable(Exception):
pass
def library_cli(spec, *args):
"""Run one inbound-library pull command; return its stdout bytes."""
root = Path(spec['root']).expanduser()
if not (root / 'inbound_library').is_dir():
raise LibraryUnavailable(f'inbound-library not found at {root}')
done = subprocess.run([sys.executable, '-m', 'inbound_library', '--root', str(root), *args],
cwd=root, capture_output=True, timeout=60)
if done.returncode:
raise LibraryUnavailable(f"inbound-library {args[0]} exited {done.returncode}")
return done.stdout
def original_date(text):
"""Send date of the forwarded original ("So., 4. Okt. 2026 um 20:07 Uhr" or
"Sun, Oct 4, 2026 at 8:07 PM"), or None."""
marker = FORWARD_MARKER.search(text or '')
match = FORWARD_DATE.search(text, marker.end()) if marker else None
if not match:
return None
value = match.group(1)
german = re.search(r'(\d{1,2})\.\s*([A-Za-zäÄ]+)\.?\s+(\d{4})', value)
english = re.search(r'([A-Za-z]{3})[a-z]*\.?\s+(\d{1,2}),?\s+(\d{4})', value)
try:
if german:
day, month, year = german.groups()
return date(int(year), MONTHS[month[:3].lower()], int(day))
if english:
month, day, year = english.groups()
return date(int(year), MONTHS[month[:3].lower()], int(day))
except (KeyError, ValueError):
return None
return None
def classify(text, received_at, zone='Europe/Berlin'):
"""Cadence and period of a mailed brief. Returns None when it is not one."""
automation = AUTOMATION.search(text or '')
if not automation:
return None
cadence = automation.group(1).lower()
sent = original_date(text)
basis = 'forwarded_date'
if sent is None:
sent = datetime.fromisoformat(received_at).astimezone(ZoneInfo(zone)).date()
basis = 'received_at'
if cadence == 'weekly':
iso = sent.isocalendar()
period = f'{iso.year}-W{iso.week:02}'
else:
period = sent.isoformat()
continued = CONTINUE.search(text + '\n')
return {'cadence': cadence, 'period': period, 'filename_date': sent.isoformat(),
'period_basis': basis, 'completeness': 'partial' if continued else 'complete',
'continue_url': continued.group(1) if continued else ''}
def pull(root, spec, *, cli=library_cli, now=None, zone='Europe/Berlin'):
"""Retain every routed brief from the configured library sources."""
root = Path(root)
moment = now or datetime.now(ZoneInfo(zone))
try:
notices = json.loads(cli(spec, 'new', '--repo', spec.get('repo', 'intelligence-radar')))
except (LibraryUnavailable, OSError, ValueError, subprocess.TimeoutExpired) as exc:
return {'ok': False, 'status': 'library_unavailable', 'error': str(exc), 'items': []}
seen = {record['content_sha256'] for _, record in all_records(root, 'external-briefs')} \
if (root / 'data' / 'external-briefs').is_dir() else set()
items = []
for notice in notices:
provider = spec.get('sources', {}).get(notice['source'])
if provider is None:
continue
item = notice['item']
if item in seen:
items.append({'item': item, 'status': 'duplicate'})
continue
try:
items.append(_retain(root, spec, cli, notice, provider, moment.isoformat(), zone))
except (LibraryUnavailable, OSError, ValueError, subprocess.TimeoutExpired) as exc:
items.append({'item': item, 'status': 'error', 'error': str(exc)})
continue
seen.add(item)
return {'ok': not any(i['status'] == 'error' for i in items), 'status': 'pulled',
'items': items}
def _retain(root, spec, cli, notice, provider, imported_at, zone):
item = notice['item']
shown = json.loads(cli(spec, 'show', item))
derived = shown.get('derived') or {}
text = derived.get('text', '')
parsed = classify(text, notice['received_at'], zone)
if parsed is None:
return {'item': item, 'status': 'not_a_brief'}
raw = cli(spec, 'get', item)
if hashlib.sha256(raw).hexdigest() != item:
raise ValueError(f'library item {item[:12]} does not match its id')
if len(raw) > MAX_BYTES:
return {'item': item, 'status': 'too_large'}
subject = derived.get('subject', '')
relative = _write_markdown(root, provider, parsed, text, item, imported_at, subject,
notice['received_at'])
atomic(root / 'data' / 'artifacts' / item, raw, immutable=True)
record = {'schema_version': 2, 'provider': provider, 'cadence': parsed['cadence'],
'period': parsed['period'], 'filename_date': parsed['filename_date'],
'source_name': subject, 'content_sha256': item, 'digest_path': relative,
'imported_at': imported_at, 'origin': 'inbound-library',
'library_source': notice['source'], 'received_at': notice['received_at'],
'period_basis': parsed['period_basis'], 'completeness': parsed['completeness'],
'continue_url': parsed['continue_url']}
return {'item': item, 'status': 'retained', 'id': put(root, 'external-briefs', record),
'digest_path': relative, 'cadence': parsed['cadence'], 'period': parsed['period'],
'completeness': parsed['completeness']}
def _write_markdown(root, provider, parsed, text, item, imported_at, subject, received_at):
folder = root / 'digests' / 'external' / parsed['cadence']
folder.mkdir(parents=True, exist_ok=True)
# The stable <period>-<provider>.md name belongs to the full brief from the drop folder.
target = folder / f"{parsed['period']}-{provider}-mail-{item[:12]}.md"
if not target.exists():
label = ('Partial: mailed teaser, cut off by the provider. The full brief needs the '
'drop folder.' if parsed['completeness'] == 'partial' else 'Complete mailed brief.')
body = '\n'.join([
'> External automation brief received by email through inbound-library. Unverified. '
'Not a reviewed digest.', f'> {label}', '',
f'- Provider: {provider}', f"- Cadence: {parsed['cadence']}",
f"- Period: {parsed['period']} (from {parsed['period_basis']})",
f'- Subject: {subject}', f'- Received: {received_at}',
f'- Library item (SHA-256 of the original message): {item}',
f"- Continue reading (needs sign-in, not fetched): {parsed['continue_url'] or '-'}",
f'- Imported: {imported_at}', '', text if text.endswith('\n') else text + '\n'])
target.write_text(body, encoding='utf-8')
return target.relative_to(root).as_posix()

9
intelligence_radar/scheduler.py
View file

@ -87,6 +87,11 @@ def _external_briefs(root, config, now):
report = import_inbox(root, resolve_inbox(spec),
archive_dirname=spec.get('archive_dirname', 'archive'), now=now)
report['native'] = native_status()
return report
except (ValueError, OSError) as exc:
return {'ok': False, 'status': 'error', 'error': str(exc), 'files': []}
report = {'ok': False, 'status': 'error', 'error': str(exc), 'files': []}
if config.get('library'):
# A library failure fails only this part of the run; source capture proceeds.
from .library_briefs import pull
report['library'] = pull(root, config['library'], now=now)
report['ok'] = report['ok'] and report['library']['ok']
return report