Masterarbeit/.claude/skills/run-experiment/extract-subagenten.py

"""Extrahiert Subagenten-Aufrufe (Prompt, Typ, Ergebnislaenge) aus dem persistierten
Session-Transkript eines Headless-Laufs und legt sie im Laufverzeichnis unter _meta ab.

Aufruf:  python subagenten.py <laufverzeichnis> [<transkript-wurzel>]
Die Session-ID wird aus <laufverzeichnis>/RawResult.json gelesen.
"""
import io, json, os, sys, glob

lauf = sys.argv[1]
wurzel = sys.argv[2] if len(sys.argv) > 2 else os.path.expandvars(r'%USERPROFILE%\.claude\projects')

roh = json.load(io.open(os.path.join(lauf, 'RawResult.json'), encoding='utf-8-sig'))
sid = roh.get('session_id')
stats = roh.get('subagent_stats') or {}
gesamt = stats.get('spawned', 0)
verschachtelt = stats.get('spawned_by_subagents', 0)
# Im Haupttranskript stehen nur die direkt vom Hauptagenten gestarteten Subagenten.
# Von Subagenten gestartete liegen in deren eigenen Transkripten.
erwartet = gesamt - verschachtelt

treffer = glob.glob(os.path.join(wurzel, '*', sid + '.jsonl'))
if not treffer:
    print('KEIN TRANSKRIPT gefunden fuer Session', sid)
    sys.exit(1)
pfad = treffer[0]

aufrufe, ergebnisse = [], {}
for zeile in io.open(pfad, encoding='utf-8'):
    try:
        o = json.loads(zeile)
    except Exception:
        continue
    cont = (o.get('message') or {}).get('content')
    if not isinstance(cont, list):
        continue
    for c in cont:
        if not isinstance(c, dict):
            continue
        if c.get('type') == 'tool_use' and c.get('name') in ('Task', 'Agent'):
            ein = c.get('input') or {}
            aufrufe.append({
                'id': c.get('id'),
                'werkzeug': c.get('name'),
                'subagent_type': ein.get('subagent_type'),
                'description': ein.get('description'),
                'run_in_background': ein.get('run_in_background'),
                'model': ein.get('model'),
                'prompt': ein.get('prompt') or '',
            })
        elif c.get('type') == 'tool_result':
            inhalt = c.get('content')
            if isinstance(inhalt, list):
                inhalt = ' '.join(str(x.get('text', '')) for x in inhalt if isinstance(x, dict))
            ergebnisse[c.get('tool_use_id')] = str(inhalt or '')

# Ein Aufruf, der am Nebenlaeufigkeitslimit scheitert, erscheint im Transkript wie ein
# regulaerer Subagenten-Start, ist aber keiner: Er liefert nur die Absage zurueck und
# zaehlt nicht in `subagent_stats.spawned`. Ohne diese Trennung meldet der Abgleich eine
# Abweichung, die es nicht gibt (Lauf Iteration 3/.../125032_v4.4.0-fb24: 21 gefundene
# Aufrufe gegenueber 13 erwarteten - die Differenz waren 8 Absagen).
ABSAGE = 'Concurrent subagent limit reached'

for a in aufrufe:
    txt = ergebnisse.get(a['id']) or ''
    a['ergebnis_zeichen'] = len(txt)
    a['abgewiesen'] = ABSAGE in txt
    # Ein im Hintergrund gestarteter Subagent liefert sofort eine Start-Quittung
    # ('Async agent launched successfully'), nicht seine Befunde. Ihre Laenge ist
    # KEIN Ertragsmass - sie ist fuer alle Hintergrund-Subagenten praktisch gleich.
    a['nur_startquittung'] = 'Async agent launched successfully' in txt

abgewiesen = [a for a in aufrufe if a['abgewiesen']]
echte = [a for a in aufrufe if not a['abgewiesen']]

meta = os.path.join(lauf, '_meta')
os.makedirs(meta, exist_ok=True)
io.open(os.path.join(meta, 'subagenten.json'), 'w', encoding='utf-8').write(
    json.dumps(aufrufe, indent=1, ensure_ascii=False))

md = ['# Subagenten-Aufrufe', '',
      'Session `%s`, Transkript `%s`.' % (sid, os.path.basename(pfad)),
      '',
      '`subagent_stats`: **%s** Subagenten gesamt, davon **%s** von Subagenten gestartet '
      '(max_depth %s). Direkt vom Hauptagenten erwartet: **%s**. Im Transkript gefunden: '
      '**%s** echte Starts%s.'
      % (gesamt, verschachtelt, stats.get('max_depth'), erwartet, len(echte),
         ' und **%d** am Nebenlaeufigkeitslimit abgewiesene Aufrufe' % len(abgewiesen)
         if abgewiesen else ''), '']
if verschachtelt:
    md += ['> Die %s von Subagenten gestarteten Aufrufe stehen in deren eigenen Transkripten und'
           ' sind hier **nicht** enthalten.' % verschachtelt, '']
if abgewiesen:
    md += ['> **%d Aufrufe wurden am Nebenlaeufigkeitslimit abgewiesen** '
           '(`subagent_stats.refused.concurrency_limit` = %s) und sind unten **nicht** '
           'aufgefuehrt. Sie erscheinen im Transkript wie regulaere Starts, liefern aber nur '
           'die Absage zurueck und zaehlen nicht in `spawned`. Fuer die Auswertung der '
           'selbstgewaehlten Zerlegung sind sie dennoch aufschlussreich: Der Hauptagent wollte '
           'staerker parallelisieren, als das Werkzeug zuliess.'
           % (len(abgewiesen), stats.get('refused', {}).get('concurrency_limit', '?')), '']
if erwartet != len(echte):
    md += ['> **Abweichung** zwischen erwarteter (%s) und gefundener Anzahl echter Starts (%s).'
           % (erwartet, len(echte)),
           '> Ursache pruefen, bevor die Prompts ausgewertet werden.', '']
for i, a in enumerate(echte, 1):
    md += ['## %d. %s' % (i, a['description'] or '(ohne Beschreibung)'),
           '',
           '- **Werkzeug:** `%s`  **Typ:** `%s`  **Hintergrund:** %s'
           % (a['werkzeug'], a['subagent_type'], a['run_in_background']),
           '- **Prompt-Zeichen:** %d  **Rueckgabe:** %s'
           % (len(a['prompt']),
              'Start-Quittung (Befunde kommen per Benachrichtigung, nicht als '
              'Werkzeugergebnis - Laenge ist kein Ertragsmass)'
              if a.get('nur_startquittung') else '%s Zeichen' % a['ergebnis_zeichen']),
           '', '### Prompt', '', '```', a['prompt'].rstrip(), '```', '']
io.open(os.path.join(meta, 'subagenten.md'), 'w', encoding='utf-8').write('\n'.join(md))

print('Subagenten gefunden: %d echte%s (direkt erwartet: %s, gesamt: %s, davon verschachtelt: %s)'
      % (len(echte), (', %d abgewiesen' % len(abgewiesen)) if abgewiesen else '',
         erwartet, gesamt, verschachtelt))
for a in echte:
    print('  - %-42s %-16s Prompt %6d Z.  %s'
          % ((a['description'] or '')[:42], a['subagent_type'], len(a['prompt']),
             'Hintergrund (Start-Quittung)' if a.get('nur_startquittung')
             else 'Ergebnis %s Z.' % a['ergebnis_zeichen']))
print('geschrieben:', os.path.join(meta, 'subagenten.md'))