Dogfood visibility: organizer pipeline runs and base-model row on the dashboard, board posts on run events, BenchFlow-native task packages accepted, challenge runs serve on four GPUs

challenges.py

@@ -36,7 +36,7 @@ CHALLENGES=[{
     'recipe':{'id':'grpo-v1','method':'GRPO (TRL) with LoRA r32/alpha64 on the policy, OpenCode rollouts in Daytona sandboxes',
         'template':'fixture/challenge-tb2-9b.toml','learning_rate':1e-6,'num_generations':8,'generation_batch_size':8,'max_steps':2,'max_completion_length':32768,
         'rollout_attempts':2,'require_reward_variance':True,'gate_task_count':32,'harness':{'agent':'opencode','skill_mode':'no-skill','concurrency':8,'agent_timeout_sec':900,'agent_idle_timeout_sec':900},
-        'sandbox':'daytona','sft':False,'teacher':False,
+        'sandbox':'daytona','sft':False,'teacher':False,'serving':{'tensor_parallel':4,'gpus':'4-7','note':'policy server on four A100s; one GPU could not keep 88 long-context tasks inside an 8 h job'},
         'note':'v1 caps training at 2 optimizer steps (64 OpenCode rollouts each) so baseline, gate, training and held-out evaluation fit one 8 h job at the measured rollout throughput (~20 model calls/min through one GPU). It proves the loop end to end; it is not enough training to expect a held-out change. A longer recipe needs cross-job checkpoint resume or more serving GPUs.'},
     'compute':{'provider':'huggingface','flavor':pipeline_jobs.FLAVOR,'timeout_seconds':8*3600,'daytona_minutes_estimate':900,'runs_per_submission_per_day':1,'concurrent_runs':1},
     'ingress':'Sandboxes reach the policy through this Space (/relay/<run>/v1) with a per-run key; the job connects outbound. No tunnel.',
@@ -208,11 +208,13 @@ def start_run(challenge_id:str,value:RunRequest,request:Request):
     if not created: return public(record)
     try:
         job=pipeline_jobs.launch(run_name=record['run_id'],config=config_path.replace('bundle/',''),bundle_rev=bundle_revision,timeout_seconds=allocation['timeout_seconds'],
-            labels={'experiment':'posttrain-challenge','challenge':challenge_id,'run_id':record['run_id']},space_origin=auth.origin(),relay_key=secrets.token_urlsafe(48))
+            labels={'experiment':'posttrain-challenge','challenge':challenge_id,'run_id':record['run_id']},space_origin=auth.origin(),relay_key=secrets.token_urlsafe(48),vllm_tp=row['recipe']['serving']['tensor_parallel'],vllm_gpus='4,5,6,7')
     except Exception:
         release(record['run_id'],'Job submission failed; reservation released.')
         raise HTTPException(502,'Job submission failed. The reservation was released; retry with the same request ID.') from None
-    return public(jobs.record_job(record,job))
+    receipt=public(jobs.record_job(record,job))
+    collab.system_post(f"Run {receipt['run_id']} started on challenge {challenge_id} for submission {source['id']} ({source.get('title','')}, {len(mirrored['tasks'])} tasks) by {user['name']}. HF job: {receipt.get('job_url')}")
+    return receipt
 
 def release(run_id,note):
     client=jobs.api()
@@ -306,7 +308,9 @@ def collect(challenge_id:str,run_id:str,request:Request):
         old=next((r for r in rows if r['run_id']==run_id),None)
         if old: return old
         rows.append(result);return result
-    return env.replace_file(RESULTS,change,[])
+    stored=env.replace_file(RESULTS,change,[])
+    collab.system_post(f"Evidence collected for run {run_id} on {challenge_id}: baseline {100*baseline['pass_rate']:.1f}% -> after {100*final['pass_rate']:.1f}% (delta {100*delta:+.1f} pp, GRPO ran: {grpo}). Awaiting organizer review.")
+    return stored
 
 @router.post('/{challenge_id}/runs/{run_id}/review')
 def review(challenge_id:str,run_id:str,value:Review,request:Request):
@@ -319,7 +323,9 @@ def review(challenge_id:str,run_id:str,value:Review,request:Request):
             if all(result.get(k)==v for k,v in update.items()): return result
             raise HTTPException(409,'Review is immutable.')
         result.update(**update,reviewed_at=now());return result
-    return env.replace_file(RESULTS,change,[])
+    reviewed=env.replace_file(RESULTS,change,[])
+    collab.system_post(f"Run {run_id} on {challenge_id} reviewed by {reviewer['name']}: {'valid, now ranked' if value.accepted else 'invalid, not ranked'}. {value.note[:300]}")
+    return reviewed
 
 def submission_rows(challenge_id):
     """Every validated submission with its latest run for this challenge (state, spend bound, milestones when running)."""
@@ -338,6 +344,19 @@ def submission_rows(challenge_id):
         rows.append(row)
     return rows
 
+def organizer_runs(limit=10):
+    """Organizer pipeline runs (phase-2 loop completion) from results/phase2-runs.json, with live job stage for open ones."""
+    try: rows=json.loads(Path(hf_hub_download(RUNS,'results/phase2-runs.json',repo_type='dataset',token=os.environ.get('HF_TOKEN'),force_download=True)).read_text())
+    except Exception: return []
+    out=[]
+    for r in rows[-limit:][::-1]:
+        item={k:r.get(k) for k in ('run_name','job_id','job_url','status','created_at','purpose','note')}
+        if item['status'] not in jobs.TERMINAL:
+            try: item['status']=str(jobs.api().inspect_job(job_id=item['job_id'],namespace='benchflow').status.stage)
+            except Exception: pass
+        out.append(item)
+    return out
+
 @router.get('/{challenge_id}/dashboard')
 def dashboard(challenge_id:str):
     """One call for the dashboard: challenge card, verified leaderboard, submissions with run state, pending results."""
@@ -352,7 +371,7 @@ def dashboard(challenge_id:str):
     card['per_run_allocation']=allocation;card['baseline']=board['baseline']
     card['participant_flow']=['python arena_cli.py environments submit --file environment.json','python arena_cli.py run --challenge '+row['id']+' --id ENVIRONMENT_ID --file run.json --execute','python arena_cli.py result collect --challenge '+row['id']+' --run-id RUN_ID']
     counts={'submissions':len(subs),'runs':sum(s['run_count'] for s in subs),'running':sum(1 for s in subs if s['state'] in ('running','queued')),'ranked':len(board['rows']),'pending_review':board['pending_count']}
-    return {'challenge':card,'challenges':[{'id':c['id'],'name':c['name'],'status':c['status']} for c in CHALLENGES],'leaderboard':board['rows'],'explanation':board['explanation'],'submissions':subs,'counts':counts,'generated_at':now()}
+    return {'challenge':card,'challenges':[{'id':c['id'],'name':c['name'],'status':c['status']} for c in CHALLENGES],'leaderboard':board['rows'],'explanation':board['explanation'],'submissions':subs,'counts':counts,'organizer_runs':organizer_runs(),'generated_at':now()}
 
 @router.get('/{challenge_id}/leaderboard')
 def leaderboard(challenge_id:str):

check_task.py

@@ -33,6 +33,7 @@ REQUIRED_FRONTMATTER = (
     "verifier",
     "environment",
 )
+BENCHFLOW_FRONTMATTER = ("task", "metadata", "agent", "verifier", "sandbox")
 REQUIRED_METADATA = (
     "author_name",
     "author_email",
@@ -74,6 +75,27 @@ def parse_metadata_keys(block: str) -> set[str]:
     return keys
 
 
+def _check_environment(task_dir: Path) -> list[str]:
+    issues: list[str] = []
+    dockerfile = task_dir / "environment" / "Dockerfile"
+    if not dockerfile.exists():
+        issues.append("Missing required file: environment/Dockerfile")
+    else:
+        # Skip comment lines and blanks; the first executable instruction
+        # must be FROM.
+        first_instr = next(
+            (
+                line
+                for line in dockerfile.read_text(encoding="utf-8").splitlines()
+                if line.strip() and not line.lstrip().startswith("#")
+            ),
+            "",
+        )
+        if not first_instr.upper().startswith("FROM "):
+            issues.append("environment/Dockerfile first instruction must be FROM")
+    return issues
+
+
 def check_task(task_dir: Path) -> list[str]:
     issues: list[str] = []
 
@@ -90,6 +112,18 @@ def check_task(task_dir: Path) -> list[str]:
         frontmatter = m.group(1)
         body = text[m.end():]
         top_keys = parse_yaml_keys(frontmatter)
+        if "schema_version" in top_keys and "task" in top_keys:
+            # BenchFlow-native task (schema 1.1): `bench tasks check` is the authority; the arena only
+            # requires the executable pieces the pipeline snapshots and runs.
+            for required in BENCHFLOW_FRONTMATTER:
+                if required not in top_keys:
+                    issues.append(f"task.md frontmatter missing: {required}")
+            if "## prompt" not in body:
+                issues.append("task.md body must contain a '## prompt' section")
+            issues.extend(_check_environment(task_dir))
+            if not (task_dir / "verifier" / "test.sh").exists():
+                issues.append("Missing required file: verifier/test.sh")
+            return issues
         for required in REQUIRED_FRONTMATTER:
             if required not in top_keys:
                 issues.append(f"task.md frontmatter missing: {required}")
@@ -100,23 +134,7 @@ def check_task(task_dir: Path) -> list[str]:
         if "## prompt" not in body:
             issues.append("task.md body must contain a '## prompt' section")
 
-    # environment/ ----------------------------------------------------------
-    dockerfile = task_dir / "environment" / "Dockerfile"
-    if not dockerfile.exists():
-        issues.append("Missing required file: environment/Dockerfile")
-    else:
-        # Skip comment lines and blanks; the first executable instruction
-        # must be FROM.
-        first_instr = next(
-            (
-                line
-                for line in dockerfile.read_text(encoding="utf-8").splitlines()
-                if line.strip() and not line.lstrip().startswith("#")
-            ),
-            "",
-        )
-        if not first_instr.upper().startswith("FROM "):
-            issues.append("environment/Dockerfile first instruction must be FROM")
+    issues.extend(_check_environment(task_dir))
 
     # verifier/ -------------------------------------------------------------
     for f in ("verifier/test.sh", "verifier/test_outputs.py", "verifier/verifier.md"):

collab.py

@@ -122,6 +122,27 @@ def post_message(value: Message, request: Request):
     result = env.replace_file(MESSAGES, change, [])
     return {'item':message_item(result),'mentions_delivered':[],'auto_subscribed':False}
 
+SYSTEM_AGENT='arena-system'
+
+def system_post(body, refs=None):
+    """Board post from the arena itself on run lifecycle events. Never raises: the main flow must not depend on the board."""
+    try:
+        def ensure_agent(rows):
+            if not any(a['agent_id']==SYSTEM_AGENT for a in rows):
+                rows.append({'agent_id':SYSTEM_AGENT,'description':'Arena system notices: run started, evidence collected, result reviewed.','model':'','harness':'posttrain-arena','owner':'benchflow','created_at':now()})
+            return rows
+        env.replace_file(AGENTS, ensure_agent, [])
+        payload={'body':body[:6000],'refs':list(refs or []),'broadcast':False}
+        row={**payload,'request_id':digest([SYSTEM_AGENT,body,int(time.time()//60)]),'owner':'benchflow','agent_id':SYSTEM_AGENT,'created_at':now(),
+             'filename':datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')+'_'+SYSTEM_AGENT+'_'+uuid.uuid4().hex[:6]+'.md','type':'agent','payload_hash':digest(payload)}
+        def change(rows):
+            if any(m.get('request_id')==row['request_id'] for m in rows): return rows
+            rows.append(row); return rows
+        env.replace_file(MESSAGES, change, [])
+        return row
+    except Exception:
+        return None
+
 class ModelRef(Strict):
     repo_id: str = Field(pattern=env.SAFE_REPO, max_length=150)
     revision: str = Field(pattern=r'^[0-9a-f]{40}$')

index.html

@@ -1249,6 +1249,14 @@
       </table>
     </div>
 
+    <div class="section-title">Organizer pipeline runs<span class="hint" id="orgRunsStatus">phase-2 loop completion on the same pipeline</span></div>
+    <div style="overflow-x:auto">
+      <table class="lb-table" style="min-width:700px">
+        <thead><tr><th style="width:260px">Run</th><th style="width:110px">Status</th><th>Note</th><th style="width:70px">Job</th></tr></thead>
+        <tbody id="orgRunsBody"></tbody>
+      </table>
+    </div>
+
     <details class="legacy-block" id="legacyBlock">
       <summary class="section-title">Practice experiments (seen-task, legacy)<span class="hint">single-task oracle-overfit runs; not held-out</span></summary>
     <div style="margin-bottom:16px"><label for="experimentGroup">Comparison group</label> <select class="btn" id="experimentGroup" aria-describedby="groupDescription"><option value="">Loading groups…</option></select><p class="subtitle" id="groupDescription">Results are ranked only within the same model, evaluation and protocol.</p></div>
@@ -4375,7 +4383,13 @@ function renderChallengeBoard(page) {
   const rows = page.leaderboard || [];
   document.getElementById('clbStatus').textContent = rows.length ? `${rows.length} ranked` : (page.counts.pending_review ? `${page.counts.pending_review} awaiting review` : 'no verified results yet');
   document.getElementById('clbExplanation').textContent = page.explanation || '';
-  if (!rows.length) { body.innerHTML = `<tr><td colspan="9" class="subtitle">No organizer-verified result yet. Submissions and their runs are listed below.</td></tr>`; return; }
+  const base = page.challenge && page.challenge.baseline;
+  if (base && base.pass_rate != null) {
+    body.insertAdjacentHTML('beforeend', `<tr class="lb-invalid-sep"><td>—</td><td><strong>${escapeHtml(page.challenge.base_model.repo_id)}</strong><div class="subtitle">base model, no post-training · organizer-measured over ${base.trials} trial${base.trials === 1 ? '' : 's'}</div></td>
+      <td class="num">${pct(base.pass_rate)}</td><td class="num">—</td><td class="num">0.0 ± ${base.stderr != null ? (100 * base.stderr).toFixed(1) : '—'}</td><td>none</td><td>—</td>
+      <td>${base.source ? `<a href="${escapeHtml(base.source)}" target="_blank" rel="noopener noreferrer">aggregate</a>` : '—'}</td><td><span class="state-badge">reference</span></td></tr>`);
+  }
+  if (!rows.length) { body.insertAdjacentHTML('beforeend', `<tr><td colspan="9" class="subtitle">No organizer-verified submission result yet. Submissions and their runs are listed below.</td></tr>`); return; }
   for (const r of rows) {
     const d = Number(r.delta_pp); const cls = d > 0 ? 'delta-pos' : '';
     body.insertAdjacentHTML('beforeend', `<tr>
@@ -4389,6 +4403,18 @@ function renderChallengeBoard(page) {
       <td><span class="state-badge published">${escapeHtml(r.verification)}</span></td></tr>`);
   }
 }
+function renderOrganizerRuns(page) {
+  const body = document.getElementById('orgRunsBody'); if (!body) return;
+  const rows = page.organizer_runs || []; body.innerHTML = '';
+  document.getElementById('orgRunsStatus').textContent = rows.length ? `${rows.length} most recent · ${rows.filter(r => r.status === 'RUNNING').length} running` : 'none recorded';
+  if (!rows.length) { body.innerHTML = `<tr><td colspan="4" class="subtitle">No organizer runs recorded.</td></tr>`; return; }
+  for (const r of rows) {
+    body.insertAdjacentHTML('beforeend', `<tr><td><strong>${escapeHtml(r.run_name || '')}</strong><div class="subtitle">${escapeHtml(String(r.created_at || '').slice(0, 16))}</div></td>
+      <td><span class="state-badge ${r.status === 'RUNNING' ? 'running' : (r.status === 'COMPLETED' ? 'published' : 'failed')}">${escapeHtml(String(r.status || '').toLowerCase())}</span></td>
+      <td class="subtitle">${escapeHtml(r.note || r.purpose || '')}</td>
+      <td>${r.job_url ? `<a href="${escapeHtml(r.job_url)}" target="_blank" rel="noopener noreferrer">job</a>` : '—'}</td></tr>`);
+  }
+}
 function renderSubmissions(page) {
   const body = document.getElementById('subsBody'); body.innerHTML = '';
   const subs = page.submissions || [];
@@ -4446,7 +4472,7 @@ async function refreshChallenge() {
     const r = await fetchWithTimeout('/api/challenges/' + encodeURIComponent(selectedChallenge) + '/dashboard');
     if (!r.ok) throw new Error('HTTP ' + r.status);
     const page = await r.json();
-    renderChallengeCard(page); renderChallengeBoard(page); renderChallengeChart(page.leaderboard || []); renderSubmissions(page);
+    renderChallengeCard(page); renderChallengeBoard(page); renderChallengeChart(page.leaderboard || []); renderSubmissions(page); renderOrganizerRuns(page);
   } catch (err) {
     document.getElementById('clbStatus').textContent = 'unavailable';
     document.getElementById('subsStatus').textContent = 'unavailable';

test_challenges.py

@@ -17,7 +17,7 @@ class ChallengeTests(unittest.TestCase):
         self.ledger={'prior_allowance_usd':50,'runs':[]};self.launches=[];self.stage='RUNNING';self.registry={challenges.RESULTS:[]}
         self.source={'id':'env-abc123abc123','challenge_id':'skillsbench','repo_type':'github','repo_id':'org/pack','revision':REV,'entry_path':'sub','title':'Pack','author':'owner','status':'Validated','source_url':'https://github.com/org/pack/tree/'+REV+'/sub'}
         self.manifest={'environment_id':self.source['id'],'revision':REV,'tasks':['alpha','beta'],'file_count':12,'bytes':1000,'mirror_revision':MIRROR}
-        self.summary=None;self.scores={}
+        self.summary=None;self.scores={};self.posts=[]
         api=SimpleNamespace(token='isolated',repo_info=lambda *a,**k:SimpleNamespace(sha=HEAD),inspect_job=lambda **k:SimpleNamespace(status=SimpleNamespace(stage=self.stage)),fetch_job_logs=lambda **k:['vllm up','bridge up','tunnel reachable'])
         patches=[patch.dict(os.environ,{'HF_TOKEN':'isolated','DAYTONA_API_KEY':'isolated'}),patch.object(jobs,'api',return_value=api),
             patch.object(jobs,'read',side_effect=lambda head=None:copy.deepcopy(self.ledger)),patch.object(jobs,'write',side_effect=self.write_ledger),
@@ -26,6 +26,7 @@ class ChallengeTests(unittest.TestCase):
             patch.object(challenges,'mirror',side_effect=lambda row:copy.deepcopy(self.manifest)),patch.object(challenges,'write_bundle',side_effect=lambda row,run_id,m:(BUNDLE,f'bundle/challenge-runs/{run_id}/config.toml')),
             patch.object(challenges,'hub',return_value=api),patch.object(challenges,'report',side_effect=lambda run_id:copy.deepcopy(self.summary)),patch.object(challenges,'stage_scores',side_effect=lambda run_id,stage,head:copy.deepcopy(self.scores[stage])),
             patch.object(challenges,'baseline_reference',return_value={'pass_rate':0.017,'stderr':0.006,'trials':2}),patch.object(pipeline_jobs,'launch',side_effect=self.launch_job),
+            patch.object(challenges,'organizer_runs',return_value=[{'run_name':'phase2-r21','job_id':'2'*24,'status':'RUNNING','note':'in flight'}]),patch.object(challenges.collab,'system_post',side_effect=lambda body,refs=None:self.posts.append(body)),
             patch.object(socket,'create_connection',side_effect=AssertionError('Unexpected network in isolated challenge test'))]
         for p in patches:p.start();self.addCleanup(p.stop)
         app=FastAPI();app.include_router(challenges.router);self.client=TestClient(app,raise_server_exceptions=False);auth._sessions.clear()
@@ -73,6 +74,7 @@ class ChallengeTests(unittest.TestCase):
         self.assertEqual(len(self.launches),1);launch=self.launches[0]
         self.assertEqual(launch['run_name'],record['run_id']);self.assertEqual(launch['config'],'challenge-runs/'+record['run_id']+'/config.toml');self.assertEqual(launch['bundle_rev'],BUNDLE);self.assertEqual(launch['timeout_seconds'],8*3600)
         self.assertTrue(launch['space_origin'].startswith('https://'));self.assertGreaterEqual(len(launch['relay_key']),48);self.assertNotIn('relay_key',json.dumps(self.ledger))
+        self.assertEqual((launch['vllm_tp'],launch['vllm_gpus']),(4,'4,5,6,7'));self.assertEqual(len(self.posts),1);self.assertIn('started on challenge tb2-9b',self.posts[0])
         ledger=self.ledger['runs'][0];self.assertEqual(ledger['kind'],'challenge-run');self.assertEqual(ledger['job_id'],'1'*24)
         self.assertEqual(self.start(),record);self.assertEqual(len(self.launches),1)
         self.post('/api/challenges/tb2-9b/runs',{'request_id':'stable-run-request','environment_id':'env-other'},expected=409)
@@ -140,5 +142,6 @@ class ChallengeTests(unittest.TestCase):
         self.post('/api/challenges/tb2-9b/runs/'+record['run_id']+'/review',{'accepted':True,'note':'Evidence reviewed: job log, score.json and per-task results agree.'},user=EDITOR)
         page=self.client.get('/api/challenges/tb2-9b/dashboard').json()
         self.assertEqual(page['submissions'][0]['state'],'published');self.assertEqual(page['leaderboard'][0]['rank'],1);self.assertEqual(page['counts']['ranked'],1)
+        self.assertEqual(page['organizer_runs'][0]['run_name'],'phase2-r21');self.assertEqual(len(self.posts),3);self.assertIn('Evidence collected',self.posts[1]);self.assertIn('reviewed by editor',self.posts[2])
 
 if __name__=='__main__':unittest.main()

test_check_task.py

@@ -0,0 +1,45 @@
+"""The structural checker accepts BenchFlow-native task packages (schema 1.1) as well as the starter-kit layout."""
+import tempfile, unittest
+from pathlib import Path
+from check_task import check_task
+
+BENCHFLOW_TASK = """---
+schema_version: '1.1'
+task:
+  name: tmax/task_x
+metadata:
+  source: legacy
+agent:
+  timeout_sec: 600.0
+verifier:
+  timeout_sec: 120.0
+sandbox:
+  cpus: 1
+---
+
+## prompt
+
+Do the thing.
+"""
+
+class CheckTaskTests(unittest.TestCase):
+    def make(self, task_md, files):
+        d = Path(tempfile.mkdtemp()) / 'task'; d.mkdir()
+        (d / 'task.md').write_text(task_md)
+        for rel, body in files.items():
+            (d / rel).parent.mkdir(parents=True, exist_ok=True); (d / rel).write_text(body)
+        return d
+    def test_benchflow_native_task_needs_only_dockerfile_and_test_sh(self):
+        d = self.make(BENCHFLOW_TASK, {'environment/Dockerfile': 'FROM python:3.12\n', 'verifier/test.sh': '#!/bin/bash\n'})
+        self.assertEqual(check_task(d), [])
+        d2 = self.make(BENCHFLOW_TASK, {'environment/Dockerfile': 'FROM python:3.12\n'})
+        self.assertIn('Missing required file: verifier/test.sh', check_task(d2))
+        d3 = self.make(BENCHFLOW_TASK.replace('sandbox:\n  cpus: 1\n', ''), {'environment/Dockerfile': 'FROM x\n', 'verifier/test.sh': ''})
+        self.assertIn('task.md frontmatter missing: sandbox', check_task(d3))
+    def test_starter_kit_layout_still_requires_oracle_and_rubrics(self):
+        kit = """---\nversion: \"1.0\"\nmetadata:\n  author_name: a\n  author_email: a@b\n  category: c\nagent:\n  timeout_sec: 1\nverifier:\n  timeout_sec: 1\nenvironment:\n  cpus: 1\n---\n\n## prompt\n"""
+        d = self.make(kit, {'environment/Dockerfile': 'FROM x\n', 'verifier/test.sh': '', 'verifier/test_outputs.py': '', 'verifier/verifier.md': ''})
+        issues = check_task(d)
+        self.assertIn('Missing required directory: verifier/rubrics/', issues); self.assertIn('Missing required file: oracle/solve.sh', issues)
+
+if __name__ == '__main__': unittest.main()