Run records: newline-only JSONL, catalogs, linked records.

JSONL splits on newlines only (a JSON string may hold U+2028, U+2029 or U+0085) and a line that does not parse is
counted, logged and reported as a finding instead of dropped silently; the search index is read the same way. The
importers' catalogs at the public root add each run's 'sampled' (which attempts were kept). Public records of the
same source run (same url) link to each other. A sampling-only run counts the tasks of its eval step. A dataset
value such as 'none (training rollouts only)' names no dataset, and an eval set a run names without logging a score
still has its dataset page.

app_api.py

@@ -275,7 +275,7 @@ def fw_run(name: str):
     pool, held = results_api.datasets_of(r)
     steps = len({row.get('step') for row in m if row.get('step') is not None}) or len(m)   # distinct steps in the whole log
     return {**r, 'metrics': fwruns.thin(m, METRIC_ROWS, lambda row: any(k in row for k in evals)), 'metrics_total': len(m), 'metrics_steps': steps, 'datasets': {'training': pool, 'heldout': held},
-            'token_ids': fwruns.token_ids(name) if r.get('steps') else []}
+            'token_ids': fwruns.token_ids(name) if r.get('steps') else [], 'linked': results_api.linked_of(r)}
 
 
 @router.get('/fwruns/{name}/metrics', response_class=PlainTextResponse)

fwruns.py

@@ -22,7 +22,7 @@ A run's steps follow AC2's vocabulary: eval@k is the held-out suite sampled afte
 groups optimizer step k+1 trained on (fw_sync rebuilds the batches from completion order; a last train@k may be a batch
 the run sampled but never trained on). Each step holds its tasks and each task its attempts (traces).
 """
-import gzip, json, math, os, re, statistics, threading, time
+import gzip, json, logging, math, os, re, statistics, threading, time
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
@@ -32,6 +32,7 @@ DEFAULTS = {'fireworks': {'source': 'fireworks', 'org': 'BenchFlow', 'project':
 PREFIX = ROOTS['fireworks']['prefix']   # the account snapshot and project reports live under the Fireworks root
 TTL = 60
 _cache, _lock = {}, threading.Lock()
+log = logging.getLogger('fwruns')
 
 
 def _repo():
@@ -137,13 +138,32 @@ def finite(v):
     return v
 
 
-def _jsonl(raw):
-    out = []
-    for line in (raw or '').splitlines():
+def _jsonl(raw, what=''):
+    """(rows, unreadable): one JSON object per line. Lines split on newlines only, since a JSON string may hold U+2028,
+    U+2029 or U+0085, which str.splitlines also splits on. A line that does not parse (a torn last line while a sync
+    writes the file, say) is counted and logged, not dropped silently."""
+    out, bad = [], 0
+    for line in (raw or '').split('\n'):
         if not line.strip(): continue
         try: out.append(finite(json.loads(line)))
-        except ValueError: continue   # a torn last line while a sync writes the file
-    return out
+        except ValueError: bad += 1
+    if bad: log.warning('%s: %d line(s) could not be read', what, bad)
+    return out, bad
+
+
+def catalog():
+    """What the importers' catalogs (_catalog*.json at the public root) say about each run, by run id: which attempts
+    were kept ('sampled') and which texts were cut. {} when there is no catalog."""
+    def load():
+        local, out = _local('public'), {}
+        names = sorted(p.name for p in local.glob('_catalog*.json')) if local and local.is_dir() else [] if local else sorted(x.path.split('/')[-1] for x in _tree('', 'public') if hasattr(x, 'size') and x.path.split('/')[-1].startswith('_catalog'))
+        for f in names:
+            try: runs_ = json.loads(_read(f, 'public') or '{}').get('runs') or []
+            except ValueError: continue
+            for r in runs_:
+                if isinstance(r, dict) and r.get('id'): out.setdefault(r['id'], {}).update({k: r[k] for k in ('sampled', 'texts_cut') if r.get(k)})
+        return out
+    return _cached('catalog', load, ROOTS['public']['ttl'])
 
 
 def _attempt(a, seen):
@@ -171,7 +191,7 @@ def runs():
             summary = {k: v for k, v in r.items() if k not in ('steps', 'metrics', 'totals')}
             training = r.get('kind', 'training') == 'training'
             # A task gives GRPO a signal only if its scored attempts differ; a screen exists to count those tasks.
-            groups = [g for s in trains for g in s['groups']]
+            groups = [g for s in (trains if training else [s for s in steps if s['kind'] == 'eval'] or trains) for g in s['groups']]   # a sampling-only run's tasks are those of its eval step
             scored = [{a['reward'] for a in g['attempts'] if a.get('reward') is not None} for g in groups]
             summary.update(steps=len(trains) if training else None, train_reward=trains[-1]['mean'] if trains and training else None, train_key=trains[-1]['key'] if trains and training else None,
                            heldout=(f"{evals[0]['mean']:.3f}" + (f" → {evals[-1]['mean']:.3f}" if len(evals) > 1 else '')) if evals else None,
@@ -208,8 +228,10 @@ def run(name):
         except ValueError: return None
         if not isinstance(meta, dict): return None
         seen = {}
-        attempts = [_attempt(a, seen) for a in _jsonl(_read(f'{name}/attempts.jsonl', root)) if isinstance(a, dict)]
-        metrics = [m for m in _jsonl(_read(f'{name}/metrics.jsonl', root)) if isinstance(m, dict)]
+        rows, bad_a = _jsonl(_read(f'{name}/attempts.jsonl', root), f'{name}/attempts.jsonl')
+        mrows, bad_m = _jsonl(_read(f'{name}/metrics.jsonl', root), f'{name}/metrics.jsonl')
+        attempts = [_attempt(a, seen) for a in rows if isinstance(a, dict)]
+        metrics = [m for m in mrows if isinstance(m, dict)]
         steps = {}
         for a in attempts:
             steps.setdefault((a['kind'], a['version']), []).append(a)
@@ -225,9 +247,10 @@ def run(name):
             ref = run(meta['baseline_run']) or {}
             ev = [st for st in ref.get('steps') or [] if st['kind'] == 'eval']
             if ev: baseline = {'run': meta['baseline_run'], **{k: v for k, v in ev[0].items() if k != 'groups'}}
-        base = {**DEFAULTS[root], **{k: v for k, v in meta.items() if v not in (None, '')}}
+        base = {**DEFAULTS[root], **({k: v for k, v in catalog().get(name, {}).items()} if root == 'public' else {}), **{k: v for k, v in meta.items() if v not in (None, '')}}
         if not base.get('project'): base['project'] = slug(base.get('org')) if base.get('org') else 'public-runs'
-        return {**meta, **base, 'id': name, 'root': root, 'project': slug(base['project']), 'steps': out, 'metrics': metrics, 'totals': _stats(attempts), 'baseline': baseline}
+        unreadable = {f: n for f, n in (('attempts.jsonl', bad_a), ('metrics.jsonl', bad_m)) if n}
+        return {**meta, **base, 'id': name, 'root': root, 'project': slug(base['project']), 'steps': out, 'metrics': metrics, 'totals': _stats(attempts), 'baseline': baseline, 'unreadable': unreadable}
     return _cached(('run', name), load, ROOTS[root]['ttl'])
 
 
@@ -326,7 +349,7 @@ def _index(name):
         raw = _read_bytes(f'{name}/search.jsonl.gz', root)
         if raw:
             out = {}
-            for line in gzip.decompress(raw).decode('utf-8', 'replace').splitlines():
+            for line in gzip.decompress(raw).decode('utf-8', 'replace').split('\n'):   # newlines only: a text may hold U+2028
                 if line.strip(): x = json.loads(line); out[x['id']] = x['parts']
             return out
         local = _local(root)

results_api.py

@@ -284,11 +284,14 @@ def as_list(v):
     return [str(x) for x in v if x] if isinstance(v, list) else []
 
 
+NONE = re.compile(r'^\s*(none|n/?a|null|no eval)\b', re.I)   # a record's way of saying it has no such dataset
+
+
 def datasets_of(r):
-    """(training data, held-out data) of a run as dataset refs {id, title}. fw_sync calls the training pool the run's
-    collection; a public record calls it dataset."""
-    pool = (as_list(r.get('collection')) or as_list(r.get('dataset')) or [None])[0]
-    held = suite_for(r.get('eval_tasks'), (as_list(r.get('eval_suite')) or [None])[0])
+    """(training data, eval set) of a run as dataset refs {id, title}. fw_sync calls the training pool the run's
+    collection; a public record calls it dataset. A value such as 'none (training rollouts only)' means there is none."""
+    pool = next((x for x in as_list(r.get('collection')) or as_list(r.get('dataset')) if not NONE.match(x)), None)
+    held = suite_for(r.get('eval_tasks'), next((x for x in as_list(r.get('eval_suite')) if not NONE.match(x)), None))
     return ({'id': slug(pool), 'title': pool} if pool else None), held
 
 
@@ -352,6 +355,15 @@ def run_item(r, snap):
             'href': run_href(r)}
 
 
+def linked_of(r):
+    """Other records of the same source run: public records with the same source URL (a run's metrics and a sample of
+    its rollouts, imported apart)."""
+    url = (r.get('url') or '').strip()
+    if not url: return []
+    return [{'id': o['id'], 'title': o.get('title') or o['id'], 'href': run_href(o), 'traced': traced(o), 'metrics_only': not traced(o) and bool(o.get('metrics'))}
+            for o in all_runs() if o['id'] != r['id'] and (o.get('url') or '').strip() == url]
+
+
 def fw_project_of_model(base_model):
     """The project of the Fireworks runs that post-train this Fireworks model resource, if any."""
     return next((r['project'] for r in all_runs() if not public(r) and r.get('base_model') == base_model), None)
@@ -535,6 +547,8 @@ def dataset_index(project='all', org='all'):
         elif held and names:   # a run that logged eval scores on its eval_suite
             s = series(r, names[0][1]); label = metric_label(names[0][0])
             add_run(ds(held, 'evaluation'), x, f"{plural(len(names), 'eval score')} logged", {'text': span(s), 'label': f'{label}, first and last logged value'})
+        elif held:   # named as the run's eval set, with no score in the records
+            add_run(ds(held, 'evaluation'), x, 'names it as its eval set', None)
     return D
 
 
@@ -1023,6 +1037,8 @@ def metric_findings(r):
 
 def findings_of(r, snap):
     F = [finding('note', 'info', 'Notes', r['note'])] if r.get('note') else []
+    for f, n in (r.get('unreadable') or {}).items():
+        F.append(finding(f'unreadable_{f}', 'medium', f"{plural(n, 'line')} of {f} could not be read", f"They are not valid JSON (a sync may have been writing the file), so they are left out of every count and chart."))
     if traced(r): F += attempt_findings(r, snap)
     if public(r) or r.get('metrics_map'): F += metric_findings(r)
     return sorted(F, key=lambda f: ORDER[f['severity']])

test_public_runs.py

@@ -279,3 +279,29 @@ def test_an_evaluations_endings_agree_with_its_pass_rate(client, roots):
     e = get(client, 'evals/fixture-agent-rollouts:eval@2').json()
     assert e['endings'] == {'passed': 2, 'failed': 2} and e['timed_out'] == {'count': 2, 'scored': 2, 'passed': 2}
     assert round(e['pass_rate'] * 4) == e['endings']['passed']
+
+
+def test_jsonl_splits_on_newlines_only_and_counts_what_it_cannot_read(roots):
+    """A JSON string may hold U+2028 or U+0085 (str.splitlines splits on them); a line that does not parse is counted."""
+    fw, pub = roots
+    d = pub / 'fixture-agent-rollouts'
+    rows = [json.loads(l) for l in (d / 'attempts.jsonl').read_text().splitlines()]
+    rows[0]['exception_message'] = 'line one
line two\u0085three'
+    (d / 'attempts.jsonl').write_text(''.join(json.dumps(a, ensure_ascii=False) + '\n' for a in rows) + '{"task": "torn')
+    fwruns._cache.clear(); R._cache.clear()
+    r = fwruns.run('fixture-agent-rollouts')
+    assert r['totals']['traces'] == len(rows) and r['unreadable'] == {'attempts.jsonl': 1}
+    assert any(a.get('exception_message') == 'line one
line two\u0085three' for s in r['steps'] for g in s['groups'] for a in g['attempts'])
+    assert {f['key'] for f in R.findings_of(r, None)} >= {'unreadable_attempts.jsonl'}
+
+
+def test_a_catalog_says_which_attempts_were_kept_and_same_source_records_link(roots, client):
+    fw, pub = roots
+    (pub / '_catalog_rollouts.json').write_text(json.dumps({'runs': [{'id': 'fixture-agent-rollouts', 'sampled': 'every rollout of steps 0 and 1'}]}))
+    write(pub, 'fixture-agent-metrics', run_json('fixture-agent-metrics', url='https://example.org/runs/fixture-agent-rollouts', eval_suite='none (training rollouts only)'), metrics_rows(4))
+    fwruns._cache.clear(); R._cache.clear()
+    assert fwruns.run('fixture-agent-rollouts')['sampled'] == 'every rollout of steps 0 and 1'
+    r = get(client, 'fwruns/fixture-agent-rollouts').json()
+    assert [x['id'] for x in r['linked']] == ['fixture-agent-metrics'] and r['linked'][0]['metrics_only']
+    m = get(client, 'fwruns/fixture-agent-metrics').json()
+    assert [x['id'] for x in m['linked']] == ['fixture-agent-rollouts'] and m['datasets']['heldout'] is None   # 'none (...)' names no eval set