Organize complete research documentation and recovery inventory

scripts/run_review26_e4b_supervisor.py

@@ -5,6 +5,11 @@ from datetime import datetime, timezone
 from pathlib import Path
 
 ROOT=Path(__file__).resolve().parents[1]
+# systemd invokes this file by absolute path, which puts ``scripts/`` rather
+# than the repository root at sys.path[0]. Runtime helpers use the ``scripts``
+# namespace, so make imports independent of the launch form and working dir.
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
 DEFAULT_RUN=ROOT/'results/phase6/review26/e4b_balanced_reward'
 
 def utc(): return datetime.now(timezone.utc).isoformat()
@@ -108,18 +113,49 @@ def start_pending_archive(run_dir,state,log,popen=subprocess.Popen):
         state['archive_worker_pid']=child.pid;state['archive_worker_step']='stop'
         _write_log(log,f'started HF stopped-run archive pid={child.pid}')
 
-def supervise(run_dir, *, poll_seconds=5, free_check=gpu_is_free, popen=subprocess.Popen):
+def supervise(run_dir, *, poll_seconds=5, free_check=gpu_is_free, popen=subprocess.Popen,
+              review28_authorized_resume=False):
     run_dir=Path(run_dir).resolve();run_dir.mkdir(parents=True,exist_ok=True)
     manifest=json.loads((run_dir/'run_manifest.json').read_text())
     if manifest.get('launch_authority')!='D-GRPO-17': raise RuntimeError('wrong launch authority')
     state_path=run_dir/'e4b_supervisor_state.json';terminal=run_dir/'e4b_terminal_stop.json'
     log=run_dir/'e4b_supervisor.log'
-    if terminal.exists(): _write_log(log,'terminal marker already exists; refusing launch');return 0
+    state=json.loads(state_path.read_text()) if state_path.exists() else dict(gpu_seconds_used=0.,restarts=0)
+    if terminal.exists():
+        if not review28_authorized_resume:
+            _write_log(log,'terminal marker already exists; refusing launch');return 0
+        try:
+            old_terminal=json.loads(terminal.read_text())
+            prior_status=old_terminal['worker_status']
+            eval_status_path=run_dir/'e4b_step100_eval_worker_status.json'
+            eval_status=json.loads(eval_status_path.read_text())
+            eval_metrics_path=Path(eval_status['metrics_path'])
+            if not eval_metrics_path.is_absolute(): eval_metrics_path=ROOT/eval_metrics_path
+        except (OSError,KeyError,ValueError,json.JSONDecodeError):
+            _write_log(log,'Review #28 resume refused: missing preserved failure/eval evidence');return 0
+        if (old_terminal.get('disposition')!='software_or_runtime_failure_hold' or
+                prior_status.get('status')!='failed' or
+                'FileExistsError' not in str(prior_status.get('stop_reason','')) or
+                eval_status.get('status')!='eval_only_complete' or
+                not eval_metrics_path.is_file() or
+                (run_dir/'e4b_supervisor_tripwire.json').exists()):
+            _write_log(log,'Review #28 resume refused: failure cause/eval result/tripwire precondition mismatch');return 0
+        if not state.get('review28_step100_eval_charged'):
+            state['gpu_seconds_used']=float(state.get('gpu_seconds_used',0.))+float(eval_status['gpu_wall_hours'])*3600
+            state['review28_step100_eval_charged']=True
+            state['review28_eval_gpu_hours']=float(eval_status['gpu_wall_hours'])
+            atomic_json(state_path,state)
+        eval_status['budget_charged_to_e4b'] = True
+        atomic_json(eval_status_path,eval_status)
+        for source,name in ((run_dir/'scale_final_status.json','scale_final_status_pre_review28.json'),
+                            (terminal,'e4b_terminal_stop_pre_review28.json')):
+            archived=run_dir/name
+            if source.exists() and not archived.exists(): os.replace(source,archived)
+        _write_log(log,'Review #28 authorized software-failure recovery: archived prior terminal markers; step-100 eval GPU time charged')
     if (run_dir/'e4b_supervisor_tripwire.json').exists():
         atomic_json(terminal,dict(status='stopped',disposition='tripwire_terminal',
             marker='e4b_supervisor_tripwire.json',at_utc=utc()))
         _write_log(log,'persisted tripwire marker found at startup; refusing auto-resume');return 0
-    state=json.loads(state_path.read_text()) if state_path.exists() else dict(gpu_seconds_used=0.,restarts=0)
     abandoned_start=state.pop('active_child_started_at',None)
     had_interrupted_child=abandoned_start is not None
     if abandoned_start is not None:
@@ -267,8 +303,9 @@ def shutil_disk_fraction(root):
 
 def main():
     ap=argparse.ArgumentParser();ap.add_argument('--run-dir',type=Path,default=DEFAULT_RUN)
+    ap.add_argument('--review28-authorized-resume',action='store_true')
     args=ap.parse_args()
-    try:return supervise(args.run_dir)
+    try:return supervise(args.run_dir,review28_authorized_resume=args.review28_authorized_resume)
     except BaseException as exc:
         run_dir=Path(args.run_dir)
         try:

scripts/run_scale_grpo.py

@@ -39,6 +39,33 @@ def select_rollout(provenance):
     return int.from_bytes(hashlib.sha256(key.encode()).digest()[:8], 'big') % 20 == 0
 
 
+def restore_e4b_eval_history(runtime, run_dir):
+    """Carry completed VALID endpoints forward when resuming a full-state run."""
+    run_dir=Path(run_dir)
+    curve_path=run_dir/'valid_curve.jsonl'
+    history=([json.loads(line) for line in curve_path.read_text().splitlines()]
+             if curve_path.is_file() else [])
+    retry_path=run_dir/'step100_matched_eval_retry/e4_valid_step_000100.json'
+    if retry_path.is_file() and not any(int(x.get('optimizer_step',-1))==100 for x in history):
+        retry=json.loads(retry_path.read_text())
+        history.append(retry)
+        append_json(curve_path,retry)
+        write_json(run_dir/'valid_step_000100.json',retry)
+    runtime.eval_entropies=[value for row in history
+        if (value:=row.get('entropy_nats',row.get('greedy_completion_entropy_nats'))) is not None]
+    for row in history:
+        score=row.get('row_auc')
+        if score is not None and score>runtime.best_p_pass:
+            runtime.best_p_pass=score
+            runtime.best_checkpoint=row.get('checkpoint')
+    if runtime.best_checkpoint is not None:
+        best_step=max((int(x.get('optimizer_step',0)) for x in history
+                       if x.get('checkpoint')==runtime.best_checkpoint),default=0)
+        write_json(run_dir/'best_checkpoint.json',dict(step=best_step,
+            valid_row_auc=runtime.best_p_pass,path=runtime.best_checkpoint,
+            selection='primary E4 endpoint row ROC-AUC on chronological VALID'))
+
+
 class ScaleRuntime(Runtime):
     sample_weight_fn=staticmethod(bounded_zero_correct_sample_difficulty_weight)
 
@@ -151,7 +178,6 @@ class ScaleRuntime(Runtime):
         ids=[r['prompt_id'] for r in self.manifest['valid_selection']]
         labels={r['prompt_id']:int(r['target']) for r in self.manifest['valid_selection']}
         prompts=[self.rows[pid]['prompt'] for pid in ids]
-        self.trainer._move_model_to_vllm()
         sampled=self.trainer.llm.generate(prompts,sampling_params=SamplingParams(
             temperature=.8,max_tokens=1536,n=8,seed=42),use_tqdm=True)
         records=[];correct_vuln_r2=[]
@@ -198,6 +224,7 @@ class ScaleRuntime(Runtime):
         if len(self.eval_entropies)>=2 and all(x<.3 for x in self.eval_entropies[-2:]):
             raise SmokeStop('two_consecutive_VALID_entropies_below_0.3 [D-GRPO-9]')
         metrics=dict(optimizer_step=step,checkpoint=str(checkpoint),adapter_label='E4 balanced reward',
+            measurement_label='E4b step-100 matched eval, post-fix retry of the interrupted measurement',
             inference_protocol='E2 exact: 8 generations, temperature 0.8, max_tokens 1536, seed 42',
             n_rows=len(records),n_pairs=len({r['pair_id'] for r in records}),
             parse_rate=sampled_parse,row_auc=score['auc'],
@@ -206,6 +233,7 @@ class ScaleRuntime(Runtime):
             greedy_label_accuracy=correct/valid_n,greedy_per_class_accuracy=per_class,
             greedy_predicted_vulnerable_fraction=predicted_vulnerable,
             greedy_completion_entropy_nats=greedy_diagnostics['entropy_nats'],
+            entropy_nats=greedy_diagnostics['entropy_nats'],
             correct_vulnerable_r2_mean=float(np.mean(correct_vuln_r2)) if correct_vuln_r2 else None,
             correct_vulnerable_r2_n=len(correct_vuln_r2),test_touched=False)
         path=self.run_dir/f'e4_valid_step_{step:06d}.json'
@@ -362,10 +390,26 @@ def write_progress_report(run_dir,status):
         axes[1,0].plot(steps,[x['completion_tokens']['mean'] for x in curves],label='mean')
         axes[1,0].plot(steps,[x['completion_tokens']['p95'] for x in curves],label='step p95')
         axes[1,0].legend();axes[1,0].set_title('Completion tokens')
-        for key in ('label_accuracy','val_p_pass_at_1'):
-            axes[1,1].plot([x['optimizer_step'] for x in validations],
-                [x[key] for x in validations],'o-',label=key)
-        axes[1,1].legend();axes[1,1].set_title('Scheduled full VALID; matched diagnostic is separate')
+        if is_e4:
+            # E4 uses the E2-matched sampled protocol. Its schema intentionally
+            # reports row_auc and greedy_label_accuracy, not the older
+            # label_accuracy / val_p_pass_at_1 pair.
+            for key, label in (('row_auc','row AUC'),
+                               ('greedy_label_accuracy','greedy accuracy')):
+                points=[x for x in validations if x.get(key) is not None]
+                if points:
+                    axes[1,1].plot([x['optimizer_step'] for x in points],
+                        [x[key] for x in points],'o-',label=label)
+            axes[1,1].set_title('E2-matched VALID AUC; greedy accuracy is secondary')
+        else:
+            for key in ('label_accuracy','val_p_pass_at_1'):
+                points=[x for x in validations if x.get(key) is not None]
+                if points:
+                    axes[1,1].plot([x['optimizer_step'] for x in points],
+                        [x[key] for x in points],'o-',label=key)
+            axes[1,1].set_title('Scheduled full VALID; matched diagnostic is separate')
+        if axes[1,1].lines:
+            axes[1,1].legend()
         for ax in axes.flat:
             ax.set_xlabel('Optimizer step')
             if manifest.get('continuation'):
@@ -375,7 +419,7 @@ def write_progress_report(run_dir,status):
         fig.tight_layout();fig.savefig(run_dir/'training_curves.png',dpi=140);plt.close(fig)
 
 
-def worker(run_dir):
+def worker(run_dir, eval_only_step=None):
     import pandas as pd
     import torch
     from datasets import Dataset
@@ -388,7 +432,7 @@ def worker(run_dir):
     from scripts.verify_grpo_reward_equivalence import sha256
 
     runtime=None;trainer=None;e4=False
-    start=float(os.environ['SCALE_START_UNIX'])
+    start=float(os.environ.get('SCALE_START_UNIX',time.time()))
     status=dict(status='initializing',started_at=now(),authorization='Review #20 + D-GRPO-14',
         test_touched=False,scale_budget_ceiling_gpu_hours=250,
         launch_segment_budget_gpu_hours=40,resume_requires_user=True)
@@ -403,7 +447,22 @@ def worker(run_dir):
         e4b=manifest.get('launch_authority')=='D-GRPO-17'
         if e4b:status['authorization']='D-GRPO-17'
         resume_file=run_dir/'e4b_resume.json'
-        if e4b and resume_file.is_file():
+        if eval_only_step is not None:
+            if not e4b or eval_only_step != 100:
+                raise SmokeStop('only_the_authorized_E4b_step100_eval_retry_is_supported')
+            from scripts.verify_grpo_reward_equivalence import sha256
+            from src.rl.full_state_checkpoint import verify_full_state
+            checkpoint=run_dir/'checkpoints'/f'step-{eval_only_step:06d}'
+            if not verify_full_state(checkpoint,expected_base_revision=manifest['policy']['base_revision']):
+                raise SmokeStop('step100_eval_checkpoint_integrity_failure')
+            adapter_hash=sha256(checkpoint/'adapter/adapter_model.safetensors')
+            continuation=dict(source_rule='D-GRPO-17',source_run=run_dir.relative_to(ROOT).as_posix(),
+                checkpoint=(checkpoint.relative_to(ROOT).as_posix()),optimizer_step=eval_only_step,
+                lr_factor=1.0,adapter_sha256=adapter_hash,adam_reset=False,scheduler_reset=False)
+            manifest['continuation']=continuation
+            status.update(authorization='Review #28 step-100 standalone matched-eval retry',
+                eval_only=True,eval_checkpoint=str(checkpoint),resume_optimizer_step=eval_only_step)
+        elif e4b and resume_file.is_file():
             continuation=json.loads(resume_file.read_text())
             if continuation.get('source_rule')!='D-GRPO-17' or continuation.get('lr_factor')!=1.0:
                 raise SmokeStop('E4b_resume_pointer_invalid')
@@ -457,8 +516,13 @@ def worker(run_dir):
             runtime.best_p_pass=resume_extra['best_p_pass']
             runtime.best_checkpoint=resume_extra['best_checkpoint']
             runtime.last_safe_step=continuation['optimizer_step']
-            prior_valid=[json.loads(x) for x in (resume_source/'valid_curve.jsonl').read_text().splitlines()]
-            runtime.eval_entropies=[x['entropy_nats'] for x in prior_valid]
+            if e4b:
+                restore_e4b_eval_history(runtime,run_dir)
+            else:
+                valid_curve_path=resume_source/'valid_curve.jsonl'
+                prior_valid=([json.loads(x) for x in valid_curve_path.read_text().splitlines()]
+                             if valid_curve_path.is_file() else [])
+                runtime.eval_entropies=[x['entropy_nats'] for x in prior_valid]
         runtime.steps_per_epoch=(len(runtime.train_ids)+63)//64
         runtime.progress('loading_fixed_judge')
         runtime.judge=FixedJudge(run_dir);runtime.judge.should_log=select_rollout
@@ -565,11 +629,22 @@ def worker(run_dir):
             out=trainer.llm.generate([rows[runtime.train_ids[0]]['prompt']],
                 sampling_params=SamplingParams(temperature=1.,max_tokens=1536,n=1),use_tqdm=False)
             record_integration_canary(runtime,trainer,out[0].outputs[0].text)
-        runtime.progress('training_scale',config_hash=config_hash,train_rows=len(runtime.train_ids),
-            valid_rows=len(manifest['valid_selection']),wall_cap_hours=40)
-        trainer.train(resume_from_checkpoint=str(bridge) if bridge else None)
-        status.update(status='stopped',stop_reason='trainer_reached_max_steps_without_review',
-            requires_user_review=True)
+        if eval_only_step is not None:
+            trainer.state.global_step=eval_only_step
+            runtime.run_dir=run_dir/'step100_matched_eval_retry'
+            runtime.run_dir.mkdir(parents=True,exist_ok=True)
+            runtime.progress('e4b_step100_matched_eval_retry',checkpoint=str(resume_checkpoint),
+                protocol='E2 exact: n=8, temperature=0.8, max_tokens=1536, seed=42')
+            metrics=runtime.evaluate_e4_matched(eval_only_step,resume_checkpoint)
+            status.update(status='eval_only_complete',metrics_path=str(runtime.run_dir/'e4_valid_step_000100.json'),
+                row_auc=metrics['row_auc'],auc_delta_vs_sft500=metrics['auc_delta_vs_sft500'],
+                sampled_parse_rate=metrics['parse_rate'],gpu_wall_hours=(time.time()-start)/3600)
+        else:
+            runtime.progress('training_scale',config_hash=config_hash,train_rows=len(runtime.train_ids),
+                valid_rows=len(manifest['valid_selection']),wall_cap_hours=40)
+            trainer.train(resume_from_checkpoint=str(bridge) if bridge else None)
+            status.update(status='stopped',stop_reason='trainer_reached_max_steps_without_review',
+                requires_user_review=True)
     except SmokeStop as error:
         status.update(status='stopped',stop_reason=str(error),requires_user_review=True)
         print('TRIPWIRE_OR_BUDGET_STOP',error,flush=True)
@@ -589,8 +664,11 @@ def worker(run_dir):
         if runtime and runtime.curve:
             status['aggregate_zero_variance_fraction']=sum(s['zero_groups'] for s in runtime.curve)/(64*len(runtime.curve))
             status['aggregate_rollout_accounting']=rollout_accounting(runtime.curve,canary=runtime.canary)
-        write_json(run_dir/'scale_final_status.json',status)
-        write_progress_report(run_dir,status)
+        status_path=(run_dir/'e4b_step100_eval_worker_status.json' if eval_only_step is not None
+                     else run_dir/'scale_final_status.json')
+        write_json(status_path,status)
+        if eval_only_step is None:
+            write_progress_report(run_dir,status)
         print('SCALE_FINAL_STATUS',json.dumps(status),flush=True)
     if status['status']=='failed':raise SystemExit(2)
 
@@ -645,10 +723,12 @@ def supervise(run_dir):
 
 def main():
     parser=argparse.ArgumentParser();parser.add_argument('--worker',action='store_true')
-    parser.add_argument('--run-dir',type=Path,required=True);args=parser.parse_args()
+    parser.add_argument('--run-dir',type=Path,required=True)
+    parser.add_argument('--eval-only-step',type=int);args=parser.parse_args()
     run_dir=args.run_dir.resolve()
     if not (run_dir/'run_manifest.json').is_file():raise SystemExit('frozen preflight manifest missing')
-    if args.worker:worker(run_dir)
+    if args.eval_only_step is not None:worker(run_dir,eval_only_step=args.eval_only_step)
+    elif args.worker:worker(run_dir)
     else:supervise(run_dir)
 
 if __name__=='__main__':main()

scripts/stage_review30_test_eval.py

@@ -0,0 +1,47 @@
+"""Stage the one-time final TEST evaluation without opening TEST data.
+
+This dry run validates the cached E4b VALID result schema and records the frozen
+four-system evaluation plan. It intentionally has no code path that reads a split
+file or invokes a model.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+VALID_FIXTURE = ROOT / "results/phase6/review26/e4b_balanced_reward/e4_valid_step_000200.json"
+OUTPUT = ROOT / "results/phase10/test_staging/dry_run.json"
+
+
+def main() -> dict:
+    record = json.loads(VALID_FIXTURE.read_text())
+    required = {"optimizer_step", "row_auc", "row_auc_pair_cluster_95_ci", "parse_rate",
+                "greedy_label_accuracy", "greedy_per_class_accuracy", "test_touched"}
+    missing = sorted(required - record.keys())
+    if missing:
+        raise ValueError(f"VALID fixture missing required metric fields: {missing}")
+    if record["test_touched"] is not False:
+        raise ValueError("VALID fixture indicates TEST was touched")
+    result = {
+        "status": "valid_fixture_schema_pass",
+        "fixture": str(VALID_FIXTURE.relative_to(ROOT)),
+        "fixture_step": record["optimizer_step"],
+        "fixture_auc": record["row_auc"],
+        "fixture_parse_rate": record["parse_rate"],
+        "metrics_schema": sorted(required),
+        "final_test_systems": ["classical pipeline", "SFT-500", "GRPO-324", "E4b step-200"],
+        "final_test_protocol": "E2-matched; accuracy + Wilson; per-class accuracy; row AUC + pair-cluster CI; IoU; P-Pass@1; trace validity",
+        "final_test_execution_authorized": False,
+        "test_split_opened": False,
+        "model_inference_run": False,
+        "note": "Dry run uses a cached chronological VALID result only; TEST path was not enumerated or read.",
+    }
+    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
+    OUTPUT.write_text(json.dumps(result, indent=2) + "\n")
+    print(json.dumps(result, indent=2))
+    return result
+
+
+if __name__ == "__main__":
+    main()

scripts/tracegrpo-e4b.service

@@ -11,7 +11,7 @@ Environment=HF_HUB_OFFLINE=1
 Environment=TOKENIZERS_PARALLELISM=false
 Environment=PYTHONUNBUFFERED=1
 ExecStartPre=+/usr/bin/systemctl stop qwen3-32b.service
-ExecStart=/home/ubuntu/ib_connect_research/project/tracegrpo/.venv/bin/python -u /home/ubuntu/ib_connect_research/project/tracegrpo/scripts/run_review26_e4b_supervisor.py
+ExecStart=/home/ubuntu/ib_connect_research/project/tracegrpo/.venv/bin/python -u -m scripts.run_review26_e4b_supervisor
 Restart=on-failure
 RestartSec=30
 KillMode=control-group

scripts/verify_complete_hf_archives.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python3
+"""Stream every recovery member and compare with the authoritative file inventory."""
+import argparse
+import hashlib
+import json
+import tarfile
+from pathlib import Path
+
+def verify(root):
+ m=json.loads((root/'SNAPSHOT_MANIFEST.json').read_text());reports=[]
+ for archive in m['archives']:
+  expected={r['path']:r for r in m['files'] if r['archive']==archive['path']};actual={}
+  h=hashlib.sha256()
+  with (root/archive['path']).open('rb') as stream:
+   for block in iter(lambda:stream.read(8*1024*1024),b''):h.update(block)
+  if h.hexdigest()!=archive['sha256']:raise ValueError('Archive SHA mismatch: '+archive['path'])
+  with tarfile.open(root/archive['path'],'r|gz') as tar:
+   for member in tar:
+    if member.name=='BUNDLE_MANIFEST.json':
+     inner=json.loads(tar.extractfile(member).read())
+     if {r['path'] for r in inner['files']}!=set(expected):raise ValueError('Bundle inventory mismatch')
+     continue
+    if member.islnk():
+     if member.linkname not in actual:raise ValueError('Missing hard-link target')
+     actual[member.name]=actual[member.linkname];continue
+    if not member.isfile():raise ValueError('Unexpected non-file archive member')
+    h=hashlib.sha256();size=0;stream=tar.extractfile(member)
+    for block in iter(lambda:stream.read(8*1024*1024),b''):h.update(block);size+=len(block)
+    actual[member.name]={'sha256':h.hexdigest(),'bytes':size}
+  if set(actual)!=set(expected):raise ValueError('Member set mismatch: '+archive['path'])
+  for path,row in expected.items():
+   if actual[path]['sha256']!=row['sha256'] or actual[path]['bytes']!=row['bytes']:raise ValueError('File content mismatch: '+path)
+  rec={'archive':archive['path'],'verified_files':len(actual),'archive_sha256':archive['sha256'],'all_member_hashes_match':True};reports.append(rec);print(json.dumps(rec),flush=True)
+ status={'status':'verified','archives':reports,'total_files':sum(r['verified_files'] for r in reports),'test_evaluated':False,'models_loaded':False}
+ (root/'ARCHIVE_VERIFICATION.json').write_text(json.dumps(status,indent=2)+'\n');return status
+
+if __name__=='__main__':
+ p=argparse.ArgumentParser();p.add_argument('release',type=Path);a=p.parse_args();verify(a.release)

src/baselines/__init__.py

@@ -0,0 +1 @@
+"""Frozen classical baselines and artifact loaders."""

src/baselines/classical_pipeline.py

@@ -0,0 +1,132 @@
+"""Persisted Phase 7 character detector and line-localizer components."""
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+import joblib
+import numpy as np
+import pandas as pd
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.linear_model import SGDClassifier
+
+from scripts.run_line_localization_baseline import materialize
+
+LABEL_CONFIG = dict(analyzer="char", ngram_range=(3, 5), min_df=2,
+                    max_features=300_000, sublinear_tf=True, dtype=np.float32)
+LABEL_MODEL_CONFIG = dict(loss="log_loss", alpha=1e-5, max_iter=200, tol=1e-3,
+                          random_state=42, average=True)
+LINE_CONFIG = dict(analyzer="char", ngram_range=(2, 5), min_df=2,
+                   max_features=250_000, sublinear_tf=True, dtype=np.float32)
+LINE_MODEL_CONFIG = dict(loss="log_loss", alpha=1e-5, max_iter=30, tol=1e-3,
+                         random_state=42, average=True, class_weight="balanced")
+ARTIFACT_FILES = {
+    "label_vectorizer": "label_vectorizer.joblib",
+    "label_classifier": "label_classifier.joblib",
+    "line_vectorizer": "line_vectorizer.joblib",
+    "line_localizer": "line_localizer.joblib",
+}
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def train_pair_texts(frame: pd.DataFrame, pair_ids: set[str]) -> tuple[list[str], np.ndarray]:
+    texts: list[str] = []
+    labels: list[int] = []
+    subset = frame[frame["pair_id"].astype(str).isin(pair_ids)]
+    for pair_id, group in subset.groupby("pair_id", sort=True):
+        if len(group) != 2 or set(group["target"].astype(int)) != {0, 1}:
+            raise ValueError(f"label pair integrity failure: {pair_id}")
+        for target in (0, 1):
+            row = group[group["target"].astype(int) == target].iloc[0]
+            texts.append(str(row["func_vuln"] if target else row["func_patch"]))
+            labels.append(target)
+    return texts, np.asarray(labels, dtype=np.int8)
+
+
+def fit_and_save(corpus: Path, train_split: Path, annotations: Path,
+                 artifact_dir: Path) -> dict[str, Any]:
+    train_ids = set(map(str, json.loads(train_split.read_text())))
+    frame = pd.read_parquet(corpus)
+    texts, labels = train_pair_texts(frame, train_ids)
+    label_vectorizer = TfidfVectorizer(**LABEL_CONFIG)
+    x_label = label_vectorizer.fit_transform(texts)
+    label_classifier = SGDClassifier(**LABEL_MODEL_CONFIG).fit(x_label, labels)
+
+    train_examples, missing, out_of_range = materialize(corpus, train_split, annotations)
+    line_docs = [candidate for ex in train_examples for candidate in ex["candidates"]]
+    line_labels = np.asarray([int(i + 1 in ex["gold"])
+                              for ex in train_examples for i in range(ex["n_lines"])], dtype=np.int8)
+    if missing or out_of_range or not line_labels.any():
+        raise ValueError(f"TRAIN localization targets invalid: missing={missing}, out_of_range={out_of_range}")
+    line_vectorizer = TfidfVectorizer(**LINE_CONFIG)
+    x_line = line_vectorizer.fit_transform(line_docs)
+    line_localizer = SGDClassifier(**LINE_MODEL_CONFIG).fit(x_line, line_labels)
+
+    artifact_dir.mkdir(parents=True, exist_ok=True)
+    objects = locals()
+    for key, filename in ARTIFACT_FILES.items():
+        joblib.dump(objects[key], artifact_dir / filename, compress=3)
+    manifest = {
+        "artifact_version": "phase7_classical_v1",
+        "fit_scope": "chronological TRAIN only",
+        "train_pairs": len(train_ids),
+        "train_rows": len(texts),
+        "train_line_candidates": len(line_labels),
+        "train_positive_lines": int(line_labels.sum()),
+        "seed": 42,
+        "label_vectorizer_config": {**LABEL_CONFIG, "dtype": "float32"},
+        "label_classifier_config": LABEL_MODEL_CONFIG,
+        "line_vectorizer_config": {**LINE_CONFIG, "dtype": "float32"},
+        "line_localizer_config": LINE_MODEL_CONFIG,
+        "versions": {"sklearn": __import__("sklearn").__version__, "numpy": np.__version__,
+                     "pandas": pd.__version__, "joblib": joblib.__version__},
+        "inputs_sha256": {p.name: sha256_file(p) for p in (corpus, train_split, annotations)},
+        "test_read": False,
+        "artifacts": {key: {"path": filename, "sha256": sha256_file(artifact_dir / filename)}
+                      for key, filename in ARTIFACT_FILES.items()},
+    }
+    (artifact_dir / "manifest.json").write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")
+    return manifest
+
+
+def load_pipeline(artifact_dir: Path, *, verify_hashes: bool = True) -> dict[str, Any]:
+    manifest = json.loads((artifact_dir / "manifest.json").read_text())
+    if manifest.get("artifact_version") != "phase7_classical_v1":
+        raise ValueError("Unsupported classical artifact version")
+    loaded = {}
+    for key, filename in ARTIFACT_FILES.items():
+        path = artifact_dir / filename
+        if verify_hashes and sha256_file(path) != manifest["artifacts"][key]["sha256"]:
+            raise ValueError(f"Artifact hash mismatch: {filename}")
+        loaded[key] = joblib.load(path)
+    loaded["manifest"] = manifest
+    return loaded
+
+
+def predict_labels(pipeline: dict[str, Any], frame: pd.DataFrame,
+                   pair_ids: set[str]) -> list[dict[str, Any]]:
+    texts, labels = train_pair_texts(frame, pair_ids)
+    subset = frame[frame["pair_id"].astype(str).isin(pair_ids)]
+    rows = []
+    for pair_id, group in subset.groupby("pair_id", sort=True):
+        for target in (0, 1):
+            row = group[group["target"].astype(int) == target].iloc[0]
+            rows.append({"sample_id": str(row["sample_id"]), "pair_id": str(pair_id), "target": target})
+    # train_pair_texts guarantees the same pair/label ordering as rows.
+    scores = pipeline["label_classifier"].decision_function(
+        pipeline["label_vectorizer"].transform(texts))
+    predictions = pipeline["label_classifier"].predict(
+        pipeline["label_vectorizer"].transform(texts)).astype(int)
+    if len(rows) != len(labels):
+        raise AssertionError("Label metadata/prediction alignment failure")
+    return [row | {"decision_score": float(score), "pred_target": int(pred)}
+            for row, score, pred in zip(rows, scores, predictions, strict=True)]

src/rl/live_adapter.py

@@ -8,7 +8,9 @@ from __future__ import annotations
 from dataclasses import dataclass
 import hashlib
 import json
+import os
 import shutil
+import tempfile
 from pathlib import Path
 
 
@@ -25,7 +27,7 @@ class AdapterSnapshot:
 
 
 def publish_live_adapter(model, root: Path, step: int, base_revision: str) -> AdapterSnapshot:
-    """Publish an immutable adapter-only copy of the current policy state."""
+    """Publish an immutable adapter copy; reuse an identical same-step publication."""
     configs = model.peft_config
     if len(configs) != 1:
         raise ValueError("D-GRPO-3 requires exactly one shared adapter")
@@ -38,21 +40,44 @@ def publish_live_adapter(model, root: Path, step: int, base_revision: str) -> Ad
         raise ValueError("adapter step and pinned base revision are required")
     if any(getattr(module, "merged", False) for module in model.modules()):
         raise ValueError("D-GRPO-1 forbids merged policy layers")
-    destination = root.resolve() / f"adapter-step-{step:06d}"
-    destination.mkdir(parents=True, exist_ok=False)
-    model.save_pretrained(destination, safe_serialization=True, selected_adapters=[name])
-    weight_file = destination / "adapter_model.safetensors"
-    if not weight_file.is_file():
-        raise ValueError("adapter-only publication has no adapter weights")
-    if any(destination.glob("model*.safetensors")):
-        raise ValueError("unexpected base/merged weights in adapter publication")
-    weights_sha = hashlib.sha256(weight_file.read_bytes()).hexdigest()
-    snapshot = AdapterSnapshot(step, destination, weights_sha, base_revision)
-    (destination / "policy_provenance.json").write_text(json.dumps({
-        "adapter_step": step, "pinned_base_revision": base_revision,
-        "adapter_weights_sha256": weights_sha, "merged": False,
-    }, indent=2) + "\n")
-    return snapshot
+    root = root.resolve()
+    root.mkdir(parents=True, exist_ok=True)
+    destination = root / f"adapter-step-{step:06d}"
+    # Serialize to a temporary directory first. If this step was already
+    # published, compare bytes and provenance before reusing it; a same-step
+    # but changed policy is a hard error, never a silent overwrite.
+    with tempfile.TemporaryDirectory(prefix=f".adapter-step-{step:06d}-", dir=root) as temp:
+        staged = Path(temp)
+        model.save_pretrained(staged, safe_serialization=True, selected_adapters=[name])
+        staged_weights = staged / "adapter_model.safetensors"
+        if not staged_weights.is_file():
+            raise ValueError("adapter-only publication has no adapter weights")
+        if any(staged.glob("model*.safetensors")):
+            raise ValueError("unexpected base/merged weights in adapter publication")
+        weights_sha = hashlib.sha256(staged_weights.read_bytes()).hexdigest()
+        provenance = {
+            "adapter_step": step, "pinned_base_revision": base_revision,
+            "adapter_weights_sha256": weights_sha, "merged": False,
+        }
+        if destination.exists():
+            weight_file = destination / "adapter_model.safetensors"
+            provenance_file = destination / "policy_provenance.json"
+            try:
+                existing = json.loads(provenance_file.read_text())
+                existing_sha = hashlib.sha256(weight_file.read_bytes()).hexdigest()
+            except (OSError, ValueError, json.JSONDecodeError) as exc:
+                raise ValueError("existing same-step adapter publication is incomplete") from exc
+            if existing != provenance or existing_sha != weights_sha:
+                raise ValueError("same-step adapter publication differs from current policy")
+            return AdapterSnapshot(step, destination, weights_sha, base_revision)
+        (staged / "policy_provenance.json").write_text(json.dumps(provenance, indent=2) + "\n")
+        try:
+            os.rename(staged, destination)
+        except FileExistsError:
+            # A concurrent publisher may have won the rename. Validate its
+            # immutable contents on the next invocation rather than overwrite.
+            raise
+        return AdapterSnapshot(step, destination, weights_sha, base_revision)
 
 
 class NativeAdapterLLM: