From 2dbe8c3a630cd43b201568a05c9732a14182fd32 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Thu, 27 Aug 2026 10:44:33 -0400 Subject: [PATCH 1/8] feat(#7303): show cron responses before run diagnostics --- api/cron_output.py | 80 +++++++ api/routes.py | 95 ++++---- static/i18n.js | 30 +++ static/panels.js | 125 +++++++--- static/style.css | 5 + tests/test_issue2289_cron_detail_expansion.py | 7 +- tests/test_issue7303_cron_response_first.py | 213 ++++++++++++++++++ tests/test_sprint10.py | 8 +- tests/test_v050257_opus_followups.py | 1 + 9 files changed, 481 insertions(+), 83 deletions(-) create mode 100644 api/cron_output.py create mode 100644 tests/test_issue7303_cron_response_first.py diff --git a/api/cron_output.py b/api/cron_output.py new file mode 100644 index 00000000000..d8e1a852b7d --- /dev/null +++ b/api/cron_output.py @@ -0,0 +1,80 @@ +"""Shared parsing and presentation projection for persisted cron output.""" + +from __future__ import annotations + +import re + + +_PREAMBLE = re.compile(r"^# Cron Job:[^\r\n]*\r?$", re.MULTILINE) +_PROMPT = re.compile(r"^## Prompt[ \t]*\r?$", re.MULTILINE) +_HEADING = re.compile(r"^#{1,2} Response[ \t]*\r?$", re.MULTILINE) +_ERROR = re.compile(r"^#{1,2} Error[ \t]*\r?$", re.MULTILINE) + + +def _outside_fence_and_quote(text: str, index: int) -> bool: + fence_char = None + fence_length = 0 + for line in text[:index].splitlines(): + stripped = line.strip() + match = re.match(r"(`{3,}|~{3,})(?:[^`~]*)$", stripped) + if not match: + continue + delimiter = match.group(1) + if fence_char is None: + fence_char, fence_length = delimiter[0], len(delimiter) + elif delimiter[0] == fence_char and len(delimiter) >= fence_length: + fence_char = None + fence_length = 0 + line_start = text.rfind("\n", 0, index) + 1 + return fence_char is None and not text[line_start:index].lstrip().startswith(">") + + +def parse_cron_output_artifact(text: str, *, job_mode: str = "unknown") -> dict: + """Return one fail-closed, raw-preserving projection for a cron artifact.""" + raw = text if isinstance(text, str) else str(text or "") + base = {"kind": "raw", "response": None, "diagnostics": raw, + "raw": raw, "fallback_reason": None} + if job_mode == "script": + base["fallback_reason"] = "script_mode" + return base + if job_mode != "agent": + base["fallback_reason"] = "unknown_mode" + return base + preamble = _PREAMBLE.match(raw) + prompt_candidates = [m for m in _PROMPT.finditer(raw, preamble.end() if preamble else 0) + if _outside_fence_and_quote(raw, m.start())] + prompt = prompt_candidates[0] if len(prompt_candidates) == 1 else None + if not preamble or not prompt or prompt.start() <= preamble.end(): + base["fallback_reason"] = "malformed_preamble" + return base + errors = [m for m in _ERROR.finditer(raw) + if _outside_fence_and_quote(raw, m.start())] + if errors: + base["fallback_reason"] = "error_output" + return base + candidates = [m for m in _HEADING.finditer(raw, prompt.end()) + if _outside_fence_and_quote(raw, m.start())] + if len(candidates) != 1: + base["fallback_reason"] = "missing_marker" if not candidates else "ambiguous_marker" + return base + marker = candidates[0] + response = raw[marker.end():].lstrip("\r\n") + if not response: + base["fallback_reason"] = "empty_response" + return base + return {"kind": "agent", "response": response, + "diagnostics": raw[:marker.start()].rstrip(), "raw": raw, + "fallback_reason": None} + + +def bounded_cron_projection(projection: dict, limit: int) -> dict: + """Bound display fields while retaining the exact raw artifact.""" + result = dict(projection) + for field in ("response", "diagnostics"): + value = result.get(field) + if isinstance(value, str) and limit >= 0 and len(value) > limit: + result[field] = value[:limit] + result[f"{field}_truncated"] = True + else: + result[f"{field}_truncated"] = False + return result diff --git a/api/routes.py b/api/routes.py index 920a3663299..b8c680fc731 100644 --- a/api/routes.py +++ b/api/routes.py @@ -1420,19 +1420,7 @@ def _is_cron_running(job_id: str) -> tuple[bool, float]: return True, time.time() - t -def _cron_response_marker_index(text: str) -> int: - """Return the start index of a markdown Response heading, if present.""" - candidates = [] - for heading in ("## Response", "# Response"): - if text.startswith(heading): - candidates.append(0) - idx = text.find(f"\n{heading}") - if idx >= 0: - candidates.append(idx + 1) - return min(candidates) if candidates else -1 - - -def _cron_output_content_window(text: str, limit: int = _CRON_OUTPUT_CONTENT_LIMIT) -> str: +def _cron_output_content_window(text: str, limit: int = _CRON_OUTPUT_CONTENT_LIMIT, *, job_mode: str = "agent") -> str: """Return a bounded cron output window that preserves useful response text. Cron output files can contain large skill dumps in the Prompt section. The @@ -1444,11 +1432,14 @@ def _cron_output_content_window(text: str, limit: int = _CRON_OUTPUT_CONTENT_LIM if len(text) <= limit: return text - response_idx = _cron_response_marker_index(text) - if response_idx >= 0: - header = text[:min(_CRON_OUTPUT_HEADER_CONTEXT, response_idx)].rstrip() - response = text[response_idx:].lstrip("\n") - content = f"{header}\n...\n{response}" if header else response + from api.cron_output import bounded_cron_projection, parse_cron_output_artifact + + projection = parse_cron_output_artifact(text, job_mode=job_mode) + bounded = bounded_cron_projection(projection, limit) + if bounded["kind"] == "agent": + header = bounded["diagnostics"][:_CRON_OUTPUT_HEADER_CONTEXT].rstrip() + response = bounded["response"] or "" + content = f"{header}\n...\n## Response\n{response}" if header else response return content[:limit] return text[-limit:] @@ -21723,9 +21714,11 @@ def _handle_cron_history(handler, parsed): for f in page: try: st = f.stat() - usage = _cron_output_usage_metadata( - f.read_text(encoding="utf-8", errors="replace") - ) + content = f.read_text(encoding="utf-8", errors="replace") + get_job = getattr(__import__("cron.jobs", fromlist=["get_job"]), "get_job", None) + job = get_job(job_id) if get_job else None + job_mode = "script" if job and job.get("no_agent") else ("agent" if job else "unknown") + usage = _cron_output_usage_metadata(content, job_mode=job_mode) runs.append({ "filename": f.name, "size": st.st_size, @@ -21760,21 +21753,35 @@ def _handle_cron_run_detail(handler, parsed): if not fpath.exists(): return j(handler, {"error": "run not found"}, status=404) try: - content = fpath.read_text(encoding="utf-8", errors="replace") - snippet = _cron_output_snippet(content) - usage = _cron_output_usage_metadata(content) + content = fpath.read_bytes().decode("utf-8", errors="replace") + import cron.jobs as cron_jobs + from api.cron_output import parse_cron_output_artifact + + get_job = getattr(cron_jobs, "get_job", None) + job = get_job(job_id) if get_job else None + job_mode = "unknown" if job is None else ("script" if job.get("no_agent") else "agent") + projection = parse_cron_output_artifact(content, job_mode=job_mode) + snippet = _cron_output_snippet(content, job_mode=job_mode) + usage = _cron_output_usage_metadata(content, job_mode=job_mode) return j(handler, {"job_id": job_id, "filename": filename, "content": content, "snippet": snippet, + "projection": projection, "usage": usage}) except Exception as e: return j(handler, {"error": str(e)}, status=500) -def _cron_output_usage_metadata(text: str) -> dict: +def _cron_output_usage_metadata(text: str, *, job_mode: str = "unknown") -> dict: """Extract optional token/cost metadata from a cron output markdown file.""" import re as _re - head = text.split("## Response", 1)[0].split("# Response", 1)[0] + head = text + if job_mode == "agent": + from api.cron_output import parse_cron_output_artifact + + projection = parse_cron_output_artifact(text, job_mode=job_mode) + if projection["kind"] == "agent": + head = projection["diagnostics"] usage: dict = {} def _intish(value: str): @@ -21789,23 +21796,23 @@ def _floatish(value: str): line = raw_line.strip() model_match = _re.match(r"\*\*(?:Model|Model Used):\*\*\s*(.+)$", line, _re.I) if model_match: - usage["model"] = model_match.group(1).strip() + usage.setdefault("model", model_match.group(1).strip()) continue provider_match = _re.match(r"\*\*Provider:\*\*\s*(.+)$", line, _re.I) if provider_match: - usage["provider"] = provider_match.group(1).strip() + usage.setdefault("provider", provider_match.group(1).strip()) continue cost_match = _re.match(r"\*\*(?:Estimated cost|Cost):\*\*\s*(.+)$", line, _re.I) if cost_match: cost = _floatish(cost_match.group(1)) if cost is not None: - usage["estimated_cost_usd"] = cost + usage.setdefault("estimated_cost_usd", cost) continue duration_match = _re.match(r"\*\*(?:Duration|Elapsed):\*\*\s*(.+)$", line, _re.I) if duration_match: seconds = _floatish(duration_match.group(1)) if seconds is not None: - usage["duration_seconds"] = seconds + usage.setdefault("duration_seconds", seconds) continue tokens_match = _re.match(r"\*\*Tokens:\*\*\s*(.+)$", line, _re.I) if tokens_match: @@ -21814,9 +21821,9 @@ def _floatish(value: str): output_match = _re.search(r"([0-9][0-9,]*)\s*(?:output|out)\b", value, _re.I) total_match = _re.search(r"([0-9][0-9,]*)\s*(?:total\s*)?tokens?\b", value, _re.I) if input_match: - usage["input_tokens"] = _intish(input_match.group(1)) + usage.setdefault("input_tokens", _intish(input_match.group(1))) if output_match: - usage["output_tokens"] = _intish(output_match.group(1)) + usage.setdefault("output_tokens", _intish(output_match.group(1))) if total_match and "total_tokens" not in usage: usage["total_tokens"] = _intish(total_match.group(1)) @@ -21827,7 +21834,7 @@ def _floatish(value: str): return usage -def _cron_output_snippet(text: str, limit: int = 600) -> str: +def _cron_output_snippet(text: str, limit: int = 600, *, job_mode: str = "agent") -> str: """Extract the response body from a cron output .md file for preview. Contract: cron output files use markdown front-matter followed by a @@ -21837,20 +21844,21 @@ def _cron_output_snippet(text: str, limit: int = 600) -> str: is returned — callers should be aware that front-matter fields (model, timestamp, …) may appear in the snippet. """ - lines = text.split("\n") - response_idx = -1 - for i, line in enumerate(lines): - if line.startswith("## Response") or line.startswith("# Response"): - response_idx = i - break - body = ("\n".join(lines[response_idx + 1:]) if response_idx >= 0 else "\n".join(lines)).strip() + from api.cron_output import bounded_cron_projection, parse_cron_output_artifact + + projection = parse_cron_output_artifact(text, job_mode=job_mode) + projection = bounded_cron_projection(projection, limit) + body = projection["response"] if projection["kind"] == "agent" else text + body = (body or "").strip() return body[:limit] or "(empty)" def _handle_cron_output(handler, parsed): - from cron.jobs import OUTPUT_DIR as CRON_OUT + import cron.jobs as cron_jobs import re as _re + CRON_OUT = cron_jobs.OUTPUT_DIR + qs = parse_qs(parsed.query) job_id = qs.get("job_id", [""])[0] if not job_id: @@ -21871,13 +21879,16 @@ def _handle_cron_output(handler, parsed): except (ValueError, TypeError): limit = 5 out_dir = CRON_OUT / job_id + get_job = getattr(cron_jobs, "get_job", None) + job = get_job(job_id) if get_job else None + job_mode = "script" if job and job.get("no_agent") else ("agent" if job else "unknown") outputs = [] if out_dir.exists(): files = sorted(out_dir.glob("*.md"), key=lambda f: f.stat().st_mtime, reverse=True)[:limit] for f in files: try: txt = f.read_text(encoding="utf-8", errors="replace") - outputs.append({"filename": f.name, "content": _cron_output_content_window(txt)}) + outputs.append({"filename": f.name, "content": _cron_output_content_window(txt, job_mode=job_mode)}) except Exception: logger.debug("Failed to read cron output file %s", f) return j(handler, {"job_id": job_id, "outputs": outputs}) diff --git a/static/i18n.js b/static/i18n.js index a5b7f9ea261..d6ceca13ace 100644 --- a/static/i18n.js +++ b/static/i18n.js @@ -1499,6 +1499,8 @@ const LOCALES = { cron_script_badge_title: 'Script job (no agent)', cron_workdir_label: 'Working directory', cron_view_full_output: 'View full output', + cron_view_diagnostics: 'View diagnostics', + cron_view_raw_output: 'View raw output', cron_all_runs: 'All runs', cron_hide_runs: 'Hide runs', cron_no_runs_yet: '(no runs yet)', @@ -3263,6 +3265,8 @@ const LOCALES = { cron_script_badge_title: 'Job script (senza agente)', cron_workdir_label: 'Directory di lavoro', cron_view_full_output: 'View full output', + cron_view_diagnostics: 'View diagnostics', + cron_view_raw_output: 'View raw output', cron_all_runs: 'Tutte le esecuzioni', cron_hide_runs: 'Nascondi esecuzioni', cron_no_runs_yet: '(nessuna esecuzione)', @@ -5027,6 +5031,8 @@ const LOCALES = { cron_script_badge_title: 'スクリプトジョブ（エージェントなし）', cron_workdir_label: '作業ディレクトリ', cron_view_full_output: '全出力を表示', + cron_view_diagnostics: '診断を表示', + cron_view_raw_output: '生の出力を表示', cron_all_runs: 'すべての実行', cron_hide_runs: '実行履歴を隠す', cron_no_runs_yet: '(まだ実行されていません)', @@ -6411,6 +6417,8 @@ const LOCALES = { cron_script_badge_title: 'Скриптовая задача (без агента)', cron_workdir_label: 'Рабочая директория', cron_view_full_output: 'Показать полный вывод', + cron_view_diagnostics: 'Показать диагностику', + cron_view_raw_output: 'Показать необработанный вывод', cron_all_runs: 'Все запуски', cron_hide_runs: 'Скрыть запуски', cron_no_runs_yet: '(пока запусков нет)', @@ -8138,6 +8146,8 @@ const LOCALES = { cron_script_badge_title: 'Script job (no agent)', cron_workdir_label: 'Working directory', cron_view_full_output: 'View full output', + cron_view_diagnostics: 'View diagnostics', + cron_view_raw_output: 'View raw output', cron_all_runs: 'All runs', cron_hide_runs: 'Hide runs', cron_no_runs_yet: '(no runs yet)', @@ -10171,6 +10181,8 @@ const LOCALES = { cron_script_path_label: 'Skriptpfad', cron_script_path_hint: 'Wird unter ~/.hermes/scripts/ aufgelöst, sofern kein absoluter Pfad. Verhalten über die Skriptdatei auf dem Server ändern.', cron_script_badge_title: 'Skript-Job (ohne Agent)', + cron_view_diagnostics: 'View diagnostics', + cron_view_raw_output: 'View raw output', cron_workdir_label: 'Arbeitsverzeichnis', cron_view_full_output: 'View full output', cron_all_runs: 'Alle Ausführungen', @@ -11552,6 +11564,8 @@ const LOCALES = { cron_script_badge_title: '脚本任务（无代理）', cron_workdir_label: '工作目录', cron_view_full_output: '查看完整输出', + cron_view_diagnostics: '查看诊断信息', + cron_view_raw_output: '查看原始输出', cron_all_runs: '全部运行记录', cron_hide_runs: '隐藏记录', cron_no_runs_yet: '（暂无运行记录）', @@ -13623,6 +13637,8 @@ const LOCALES = { cron_script_badge_title: '腳本任務（無代理）', cron_workdir_label: '工作目錄', cron_view_full_output: '檢視完整輸出', + cron_view_diagnostics: '檢視診斷資訊', + cron_view_raw_output: '檢視原始輸出', cron_all_runs: '所有執行', cron_hide_runs: '隱藏執行記錄', cron_no_runs_yet: '（尚無執行記錄）', @@ -15199,6 +15215,8 @@ const LOCALES = { cron_script_path_label: 'Caminho do script', cron_script_path_hint: 'Resolvido em ~/.hermes/scripts/ salvo caminho absoluto. Edite o script no servidor para alterar o comportamento.', cron_script_badge_title: 'Tarefa script (sem agente)', + cron_view_diagnostics: 'View diagnostics', + cron_view_raw_output: 'View raw output', cron_workdir_label: 'Diretório de trabalho', cron_view_full_output: 'View full output', cron_all_runs: 'Todas execuções', @@ -16857,6 +16875,8 @@ const LOCALES = { cron_script_badge_title: 'Script job (no agent)', cron_workdir_label: 'Working directory', cron_view_full_output: 'View full output', + cron_view_diagnostics: 'View diagnostics', + cron_view_raw_output: 'View raw output', cron_all_runs: 'All runs', cron_hide_runs: 'Hide runs', cron_no_runs_yet: '(no runs yet)', @@ -18693,6 +18713,8 @@ const LOCALES = { cron_script_badge_title: 'Tâche script (sans agent)', cron_workdir_label: 'Répertoire de travail', cron_view_full_output: 'Voir la sortie complète', + cron_view_diagnostics: 'Voir les diagnostics', + cron_view_raw_output: 'Voir la sortie brute', cron_all_runs: 'Toutes les exécutions', cron_hide_runs: 'Masquer les exécutions', cron_no_runs_yet: '(pas encore d\'exécutions)', @@ -19141,6 +19163,8 @@ const LOCALES = { cron_toast_notifications_hint: 'Zobrazit toast po dokončení tohoto cronu. Odznak Úloh a nový ukazatel běhu se stále aktualizují když je toto vypnuto.', cron_toast_notifications_label: 'Dokončovací notifikace', cron_view_full_output: 'Zobrazit celý výstup', + cron_view_diagnostics: 'Zobrazit diagnostiku', + cron_view_raw_output: 'Zobrazit nezpracovaný výstup', cron_workdir_label: 'Pracovní adresář', csv_error: 'Nepodařilo se načíst CSV soubor', csv_header_note: 'První řádek zobrazen jako záhlaví tabulky', @@ -22071,6 +22095,8 @@ const LOCALES = { cron_script_path_label: 'Betik yolu', cron_script_path_hint: 'Mutlak yol değilse ~/.hermes/scripts/ altında çözülür. Davranışı değiştirmek için sunucudaki betiği düzenleyin.', cron_script_badge_title: 'Betik işi (ajan yok)', + cron_view_diagnostics: 'View diagnostics', + cron_view_raw_output: 'View raw output', cron_workdir_label: 'Çalışma dizini', cron_view_full_output: 'View full output', cron_all_runs: 'Tüm koşular', @@ -23917,6 +23943,8 @@ const LOCALES = { cron_script_badge_title: 'Zadanie skryptowe (bez agenta)', cron_workdir_label: 'Katalog roboczy', cron_view_full_output: 'Wyświetl pełne wyjście', + cron_view_diagnostics: 'Wyświetl diagnostykę', + cron_view_raw_output: 'Wyświetl surowe dane', cron_all_runs: 'Wszystkie uruchomienia', cron_hide_runs: 'Ukryj uruchomienia', cron_no_runs_yet: '(brak uruchomień)', @@ -25506,6 +25534,8 @@ const LOCALES = { cron_script_badge_title: 'Job script (không dùng agent)', cron_workdir_label: 'Thư mục làm việc', cron_view_full_output: 'Xem toàn bộ đầu ra', + cron_view_diagnostics: 'Xem chẩn đoán', + cron_view_raw_output: 'Xem đầu ra thô', cron_all_runs: 'Tất cả lần chạy', cron_hide_runs: 'Ẩn lần chạy', cron_no_runs_yet: '(chưa có lần chạy nào)', diff --git a/static/panels.js b/static/panels.js index 2e9acca4ed7..240cc45eb78 100644 --- a/static/panels.js +++ b/static/panels.js @@ -1151,6 +1151,15 @@ function _cronExpansionSet(key, expanded){ try { localStorage.setItem(key, expanded ? '1' : '0'); } catch(_) {} } +function _syncCronRunExpandControl(item, expanded){ + const btn = item ? item.querySelector('.detail-expand-toggle') : null; + if (!btn) return; + btn.textContent = expanded ? '▴' : '▾'; + btn.title = expanded ? (t('cron_collapse_output') || 'Collapse output') : (t('cron_expand_output') || 'Expand output'); + btn.setAttribute('aria-label', btn.title); + btn.setAttribute('aria-expanded', String(expanded)); +} + function toggleCronPromptExpanded(jobId){ const key = _cronPanelExpandKey(jobId, 'prompt'); _cronExpansionSet(key, !_cronExpansionGet(key)); @@ -1161,17 +1170,25 @@ function toggleCronPromptExpanded(jobId){ function toggleCronRunExpanded(jobId, filename, runId){ const key = _cronRunExpandKey(jobId, filename); + const item = document.getElementById(runId); + if (item && !item.classList.contains('open')) { + _cronExpansionSet(key, true); + _loadRunContent(jobId, filename, runId); + return; + } const expanded = !_cronExpansionGet(key); _cronExpansionSet(key, expanded); - const item = document.getElementById(runId); const body = item ? item.querySelector('.detail-run-body') : null; - const btn = item ? item.querySelector('.detail-expand-toggle') : null; if (body) body.classList.toggle('expanded', expanded); - if (btn) { - btn.textContent = expanded ? '▴' : '▾'; - btn.title = expanded ? (t('cron_collapse_output') || 'Collapse output') : (t('cron_expand_output') || 'Expand output'); - btn.setAttribute('aria-label', btn.title); + if (body) { + const primary = body.querySelector('.cron-run-primary code'); + if (primary && body.dataset.fullResponse !== undefined) { + primary.textContent = expanded ? body.dataset.fullResponse : body.dataset.fullResponse.slice(0, 600); + } } + _syncCronRunExpandControl(item, expanded); + const btn = item ? item.querySelector('.detail-expand-toggle') : null; + if (btn) btn.focus({preventScroll:true}); } function _isCronScriptJob(job){ @@ -1393,9 +1410,8 @@ async function _loadRunContent(jobId, filename, runId){ _cronExpansionSet(_cronRunExpandKey(jobId, filename), false); const btn = item ? item.querySelector('.detail-expand-toggle') : null; if (btn) { - btn.textContent = '▾'; - btn.title = (t('cron_expand_output') || 'Expand output'); - btn.setAttribute('aria-label', btn.title); + _syncCronRunExpandControl(item, false); + btn.focus({preventScroll:true}); } return; } @@ -1409,17 +1425,36 @@ async function _loadRunContent(jobId, filename, runId){ return; } const expanded = _cronExpansionGet(_cronRunExpandKey(jobId, filename)); - const output = expanded ? (data.content || data.snippet || '') : (data.snippet || data.content || ''); + const projection = data.projection || {kind:'raw', raw:data.content || ''}; + const isResponse = projection.kind === 'agent' && typeof projection.response === 'string'; + const fullResponse = isResponse ? projection.response : (projection.raw || data.content || ''); + let renderedOutput; + let expandableOutput = fullResponse; + if (isResponse) { + renderedOutput = expanded ? fullResponse : fullResponse.slice(0, 600); + } else { + // Keep the legacy selection for old payloads, then cap what reaches the DOM. + const output = expanded ? (data.content || data.snippet || '') : (data.snippet || data.content || ''); + const fullFallbackOutput = data.content || fullResponse || data.snippet || ''; + expandableOutput = fullFallbackOutput; + renderedOutput = expanded ? fullFallbackOutput : output.slice(0, 600); + } + let hasLegacyFallbackExpansion = false; + if (!expanded && data.content && data.snippet && data.content.length > data.snippet.length) { + hasLegacyFallbackExpansion = true; + } + body.dataset.fullResponse = expandableOutput; body.classList.toggle('expanded', expanded); + _syncCronRunExpandControl(item, expanded); // Cron run output is never authored Markdown — render as literal // preformatted text using DOM-created 
 so all content
     // (including shapes starting with #, |, >, ``` and embedded fences)
     // renders verbatim without Markdown interpretation.
     body.innerHTML = '';
     const pre = document.createElement('pre');
-    pre.className = 'cron-run-pre';
+    pre.className = 'cron-run-pre cron-run-primary';
     const code = document.createElement('code');
-    code.textContent = output;
+    code.textContent = renderedOutput;
     pre.appendChild(code);
     body.appendChild(pre);
     const usageStrip = _formatCronRunUsageStrip(data.usage);
@@ -1429,28 +1464,52 @@ async function _loadRunContent(jobId, filename, runId){
       usage.textContent = usageStrip;
       body.appendChild(usage);
     }
-    // Show "View full output" button only for collapsed previews. Expanded rows render the full body inline.
-    if (!expanded && data.content && data.snippet && data.content.length > data.snippet.length) {
+    if (isResponse && projection.diagnostics) {
+      const diagnostics = document.createElement('details');
+      diagnostics.className = 'cron-run-diagnostics';
+      const summary = document.createElement('summary');
+      summary.textContent = t('cron_view_diagnostics') || 'View diagnostics';
+      diagnostics.append(summary);
+      diagnostics.addEventListener('toggle', () => {
+        if (!diagnostics.open || diagnostics.dataset.loaded) return;
+        const diagnosticPre = document.createElement('pre');
+        diagnosticPre.className = 'cron-run-pre cron-run-diagnostics-content';
+        diagnosticPre.textContent = projection.diagnostics;
+        diagnostics.appendChild(diagnosticPre);
+        diagnostics.dataset.loaded = '1';
+      });
+      body.appendChild(diagnostics);
+    }
+    if (isResponse) {
+      const rawDetails = document.createElement('details');
+      rawDetails.className = 'cron-run-raw-output';
+      const rawSummary = document.createElement('summary');
+      rawSummary.textContent = t('cron_view_raw_output') || 'View raw output';
+      rawDetails.append(rawSummary);
+      rawDetails.addEventListener('toggle', () => {
+        if (!rawDetails.open || rawDetails.dataset.loaded) return;
+        const raw = document.createElement('pre');
+        raw.className = 'cron-run-pre';
+        raw.textContent = data.content || '';
+        rawDetails.appendChild(raw);
+        rawDetails.dataset.loaded = '1';
+      });
+      body.appendChild(rawDetails);
+    }
+    // Expansion changes only the response cap; raw output remains a separate action.
+    if (!expanded && ((isResponse && fullResponse.length > 600) || hasLegacyFallbackExpansion)) {
       const btn = document.createElement('button');
-      btn.style.cssText = 'margin-top:8px;padding:4px 12px;border-radius:var(--radius-btn);border:1px solid var(--border-subtle);background:var(--surface-subtle);color:var(--text-secondary);cursor:pointer;font-size:12px';
+      btn.type = 'button';
+      btn.className = 'cron-run-response-toggle';
       btn.textContent = t('cron_view_full_output') || 'View full output';
       btn.onclick = () => {
         _cronExpansionSet(_cronRunExpandKey(jobId, filename), true);
         body.classList.add('expanded');
-        body.innerHTML = '';
-        const pre = document.createElement('pre');
-        pre.className = 'cron-run-pre';
-        const code = document.createElement('code');
-        code.textContent = data.content || '';
-        pre.appendChild(code);
-        body.appendChild(pre);
-        const usageStrip = _formatCronRunUsageStrip(data.usage);
-        if (usageStrip) {
-          const usage = document.createElement('div');
-          usage.className = 'cron-run-usage-strip cron-run-usage-footer';
-          usage.textContent = usageStrip;
-          body.appendChild(usage);
-        }
+        const primary = body.querySelector('.cron-run-primary code');
+        if (primary) primary.textContent = expandableOutput;
+        _syncCronRunExpandControl(item, true);
+        const toggle = item.querySelector('.detail-expand-toggle');
+        if (toggle) toggle.focus({preventScroll:true});
         btn.remove();
       };
       body.appendChild(btn);
@@ -1985,14 +2044,6 @@ async function saveCronForm(){
 const submitCronCreate = saveCronForm;
 function toggleCronForm(){ openCronCreate(); }
 
-function _cronOutputSnippet(content) {
-  // Extract the response body from a cron output .md file
-  const lines = content.split('\n');
-  const responseIdx = lines.findIndex(l => l.startsWith('## Response') || l.startsWith('# Response'));
-  const body = (responseIdx >= 0 ? lines.slice(responseIdx + 1) : lines).join('\n').trim();
-  return body.slice(0, 600) || '(empty)';
-}
-
 function _formatCronRunUsageStrip(usage) {
   if (!usage || typeof usage !== 'object') return '';
   const parts = [];
diff --git a/static/style.css b/static/style.css
index 98f6b7f0fec..e6b9998b42c 100644
--- a/static/style.css
+++ b/static/style.css
@@ -6485,6 +6485,11 @@ main.main > .main-view:not([id="mainChat"]):not([id="mainSettings"]) .main-view-
 .detail-run-body.expanded{max-height:none;overflow-y:visible;}
 .detail-run-body .cron-run-pre{margin:0;font-size:12px;line-height:1.5;white-space:pre-wrap;word-break:break-word;}
 .detail-run-body .cron-run-pre code{background:transparent;padding:0;font-family:var(--font-mono);}
+.cron-run-diagnostics{margin-top:8px;border-top:1px solid var(--border);padding-top:6px;color:var(--muted);}
+.cron-run-diagnostics summary{cursor:pointer;overflow-wrap:anywhere;}
+.cron-run-diagnostics-content,.cron-run-raw-output pre{margin-top:6px;max-height:220px;overflow:auto;}
+.cron-run-raw-output summary,.cron-run-response-toggle{margin-top:8px;padding:4px 12px;border:1px solid var(--border-subtle);border-radius:var(--radius-btn);background:var(--surface-subtle);color:var(--text-secondary);cursor:pointer;font-size:12px;max-width:100%;white-space:normal;text-align:left;}
+@container rightpanel (max-width:520px){.detail-run-actions{gap:4px}.cron-run-response-toggle{width:100%;}.cron-run-diagnostics summary,.cron-run-raw-output summary{line-height:1.4;}}
 .workspace-panel-tabs{display:flex;gap:4px;padding:6px 8px;border-bottom:1px solid var(--border);}
 .workspace-panel-tab{flex:1;border:1px solid transparent;background:transparent;color:var(--muted);border-radius:7px;padding:5px 8px;font-size:12px;cursor:pointer;}
 .workspace-panel-tab.active{background:var(--surface-subtle);color:var(--text);border-color:var(--border2);}
diff --git a/tests/test_issue2289_cron_detail_expansion.py b/tests/test_issue2289_cron_detail_expansion.py
index d3d83ee2efb..9fadf9edf8c 100644
--- a/tests/test_issue2289_cron_detail_expansion.py
+++ b/tests/test_issue2289_cron_detail_expansion.py
@@ -41,5 +41,10 @@ def test_cron_expansion_i18n_keys_exist_in_every_locale():
         "cron_collapse_prompt",
         "cron_expand_output",
         "cron_collapse_output",
+        "cron_view_diagnostics",
+        "cron_view_raw_output",
     ):
-        assert I18N_JS.count(f"{key}:") >= locale_count
+        assert I18N_JS.count(f"{key}:") == locale_count
+    english = I18N_JS.split("  en: {", 1)[1].split("\n  it: {", 1)[0]
+    assert english.count("cron_view_diagnostics:") == 1
+    assert english.count("cron_view_raw_output:") == 1
diff --git a/tests/test_issue7303_cron_response_first.py b/tests/test_issue7303_cron_response_first.py
new file mode 100644
index 00000000000..4dc32a5eb11
--- /dev/null
+++ b/tests/test_issue7303_cron_response_first.py
@@ -0,0 +1,213 @@
+"""Behavioral regressions for cron response-first run details."""
+
+from __future__ import annotations
+
+import io
+import json
+import types
+from pathlib import Path
+
+import pytest
+
+
+def agent_artifact(response="Done", *, newline="\n"):
+    return newline.join(("# Cron Job: demo", "", "**Job ID:** abc", "", "## Prompt", "", "context", "", "## Response", "", response))
+
+
+def large_issue_artifact():
+    prefix = ["# Cron Job: Night", "", "**Job ID:** abc1", "", "## Prompt"]
+    prefix.extend(f"prompt context line {i:03d} " + ("x" * 114) for i in range(315))
+    response = ["## Response"] + [f"useful response line {i:02d}" for i in range(24)]
+    raw = "\n".join(prefix + response)
+    assert len(raw.encode("utf-8")) == 44419
+    assert len(raw.splitlines()) == 345
+    return raw
+
+
+def test_parser_accepts_canonical_crlf_and_preserves_raw():
+    from api.cron_output import parse_cron_output_artifact
+    raw = agent_artifact("Done\r\n✓", newline="\r\n")
+    projection = parse_cron_output_artifact(raw, job_mode="agent")
+    assert projection["kind"] == "agent"
+    assert projection["response"] == "Done\r\n✓"
+    assert projection["diagnostics"].endswith("## Prompt\r\n\r\ncontext")
+    assert projection["raw"] == raw
+
+
+def test_parser_preserves_leading_response_indentation_and_ignores_fenced_prompt_error():
+    from api.cron_output import parse_cron_output_artifact
+
+    raw = agent_artifact("  indented\n    continuation", newline="\r\n").replace(
+        "## Prompt\r\n\r\ncontext", "## Prompt\r\n\r\n```\r\n## Error\r\n```\r\ncontext"
+    ).replace("# Cron Job: demo\r\n\r\n## Prompt", "# Cron Job: demo\r\n\r\n```\r\n## Prompt\r\n```\r\n## Prompt")
+    projection = parse_cron_output_artifact(raw, job_mode="agent")
+    assert projection["kind"] == "agent"
+    assert projection["response"].startswith("  indented\n")
+
+
+def test_parser_fails_closed_for_script_unknown_and_ambiguous_inputs():
+    from api.cron_output import parse_cron_output_artifact
+    raw = agent_artifact()
+    for mode in ("script", "unknown"):
+        result = parse_cron_output_artifact(raw, job_mode=mode)
+        assert result["kind"] == "raw" and result["raw"] == raw
+    fenced = agent_artifact("```\n## Response\n```\n\n## Response\nreal")
+    assert parse_cron_output_artifact(fenced, job_mode="agent")["kind"] == "raw"
+    ambiguous = agent_artifact("first\n\n## Response\nsecond\n\n# Response\nthird")
+    assert parse_cron_output_artifact(ambiguous, job_mode="agent")["fallback_reason"] == "ambiguous_marker"
+
+
+def test_error_marker_wins_over_response_marker_in_prompt_text():
+    from api.cron_output import parse_cron_output_artifact
+
+    raw = agent_artifact().replace(
+        "context", "context\n\n## Response\nquoted prompt example\n\n## Error\n\n`failed`"
+    )
+    result = parse_cron_output_artifact(raw, job_mode="agent")
+    assert result["kind"] == "raw"
+    assert result["fallback_reason"] == "error_output"
+    assert result["raw"] == raw
+
+
+def test_mixed_fences_keep_inner_tilde_fence_closed():
+    from api.cron_output import parse_cron_output_artifact
+
+    raw = agent_artifact("").split("## Response", 1)[0] + "```\n~~~\n## Response\n~~~\n```\n\n## Response\n\nreal"
+    result = parse_cron_output_artifact(raw, job_mode="agent")
+    assert result["kind"] == "agent"
+    assert result["response"] == "real"
+
+
+def test_response_headings_cannot_overwrite_route_usage_metadata():
+    from api.routes import _cron_output_usage_metadata
+
+    raw = agent_artifact("**Model Used:** fake\n**Cost:** $999\n**Tokens:** 1 input, 2 output")
+    raw = raw.replace("**Job ID:** abc", "**Job ID:** abc\n**Provider:** route-provider\n**Model Used:** route-model\n**Cost:** $0.12\n**Duration:** 4.5s\n**Tokens:** 100 input, 20 output")
+    raw = raw.replace("**Model Used:** fake", "**Provider:** fake-provider\n**Model Used:** fake")
+    assert _cron_output_usage_metadata(raw, job_mode="agent") == {
+        "provider": "route-provider",
+        "model": "route-model", "estimated_cost_usd": 0.12,
+        "duration_seconds": 4.5, "input_tokens": 100, "output_tokens": 20, "total_tokens": 120,
+    }
+
+
+def test_large_prompt_reproduction_preserves_response_in_bounded_window():
+    from api.routes import _cron_output_content_window
+
+    raw = large_issue_artifact()
+    projection = __import__("api.cron_output", fromlist=["parse_cron_output_artifact"]).parse_cron_output_artifact(raw, job_mode="agent")
+    window = _cron_output_content_window(raw, limit=8000, job_mode="agent")
+    assert projection["response"].startswith("useful response line 00")
+    assert "useful response line 23" in window
+    assert len(window) <= 8000
+
+
+def test_parser_fallbacks_keep_error_and_empty_artifacts_raw():
+    from api.cron_output import parse_cron_output_artifact
+    for suffix, reason in (("\r\n\r\n## Error\r\n\r\n`oops`", "error_output"), ("\r\n\r\n## Response\r\n", "empty_response")):
+        raw = agent_artifact().split("## Response", 1)[0] + suffix
+        result = parse_cron_output_artifact(raw, job_mode="agent")
+        assert result["kind"] == "raw" and result["fallback_reason"] == reason
+
+
+class _Handler:
+    def __init__(self): self.status, self.wfile = None, io.BytesIO()
+    def send_response(self, status): self.status = status
+    def send_header(self, *_): pass
+    def end_headers(self): pass
+
+
+def test_run_detail_returns_shared_projection_and_exact_content(monkeypatch, tmp_path):
+    import api.routes as routes
+    output = tmp_path / "job_abc" / "run.md"
+    output.parent.mkdir()
+    raw = agent_artifact("literal response", newline="\r\n")
+    output.write_bytes(raw.encode())
+    jobs = types.ModuleType("cron.jobs")
+    jobs.OUTPUT_DIR = tmp_path
+    jobs.get_job = lambda job_id: {"id": job_id, "no_agent": False}
+    cron = types.ModuleType("cron"); cron.__path__ = []
+    monkeypatch.setitem(__import__("sys").modules, "cron", cron)
+    monkeypatch.setitem(__import__("sys").modules, "cron.jobs", jobs)
+    handler = _Handler()
+    routes._handle_cron_run_detail(handler, types.SimpleNamespace(query="job_id=job_abc&filename=run.md"))
+    body = json.loads(handler.wfile.getvalue())
+    assert handler.status == 200 and body["content"] == raw
+    assert body["projection"]["response"] == "literal response"
+    assert body["projection"]["raw"] == raw
+
+
+def _load_run_function():
+    return _extract_function("_loadRunContent", async_function=True)
+
+
+def _extract_function(name, *, async_function=False):
+    source = (Path(__file__).parents[1] / "static" / "panels.js").read_text(encoding="utf-8")
+    prefix = "async function " if async_function else "function "
+    start = source.index(prefix + name + "(")
+    depth = 0
+    for index in range(source.index("{", start), len(source)):
+        depth += source[index] == "{"
+        depth -= source[index] == "}"
+        if depth == 0: return source[start : index + 1]
+    raise AssertionError("could not extract run-detail loader")
+
+
+def test_run_detail_dom_keeps_response_primary_and_raw_separate():
+    playwright = pytest.importorskip("playwright.sync_api")
+    with playwright.sync_playwright() as pw:
+        browser = pw.chromium.launch(headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"])
+        page = browser.new_page()
+        page.set_content('
')
+        page.add_script_tag(content="""
+            window.t = key => ({cron_view_diagnostics:'View diagnostics', cron_view_raw_output:'View raw output', cron_view_full_output:'View full output'})[key] || key;
+            window.esc = value => String(value);
+            window.api = async () => ({projection:{kind:'agent',response:'VISIBLE RESPONSE',diagnostics:'PROMPT CONTEXT',raw:'RAW ARTIFACT'},content:'RAW ARTIFACT'});
+            const expansions = {};
+            window._cronExpansionGet = key => !!expansions[key];
+            window._cronExpansionSet = (key, value) => { expansions[key] = !!value; };
+            window._cronRunExpandKey = () => 'run'; window._formatCronRunUsageStrip = () => '';
+        """)
+        page.add_script_tag(content=_extract_function("_syncCronRunExpandControl"))
+        page.add_script_tag(content=_extract_function("toggleCronRunExpanded"))
+        page.add_script_tag(content=_load_run_function())
+        page.evaluate("_loadRunContent('job','run.md','run')")
+        page.wait_for_selector(".cron-run-primary")
+        assert page.locator(".cron-run-primary").inner_text() == "VISIBLE RESPONSE"
+        assert page.locator(".cron-run-diagnostics").count() == 1
+        assert page.locator(".cron-run-raw-output").count() == 1
+        assert page.locator(".cron-run-raw-output pre").count() == 0
+        assert page.locator(".detail-expand-toggle").get_attribute("aria-expanded") == "false"
+        page.locator(".detail-expand-toggle").click()
+        assert page.locator(".detail-expand-toggle").evaluate("el => document.activeElement === el")
+        assert page.locator(".detail-expand-toggle").get_attribute("aria-expanded") == "true"
+        assert page.locator(".cron-run-primary").inner_text() == "VISIBLE RESPONSE"
+        page.locator(".cron-run-diagnostics summary").click()
+        page.wait_for_selector(".cron-run-diagnostics-content")
+        assert page.locator(".cron-run-diagnostics-content").count() == 1
+        page.locator(".cron-run-raw-output summary").click()
+        page.wait_for_selector(".cron-run-raw-output pre")
+        assert page.locator(".cron-run-raw-output pre").count() == 1
+        page.locator(".detail-expand-toggle").click()
+        assert page.locator(".cron-run-primary").inner_text() == "VISIBLE RESPONSE"
+
+        page.evaluate("window.api = async () => ({projection:{kind:'raw',raw:'RAW ARTIFACT'},content:'RAW ARTIFACT'})")
+        page.evaluate("document.body.insertAdjacentHTML('beforeend', '
')")
+        page.evaluate("_loadRunContent('job','run2.md','run2')")
+        page.wait_for_selector("#run2 .cron-run-primary")
+        assert page.locator("#run2 .cron-run-primary").inner_text() == "RAW ARTIFACT"
+        assert page.locator("#run2 .cron-run-raw-output").count() == 0
+
+        page.evaluate("""
+            window.api = async () => {
+                const content = 'x'.repeat(900);
+                return {projection:{kind:'raw',raw:content},content,snippet:content.slice(0, 600)};
+            };
+            document.body.insertAdjacentHTML('beforeend', '
');
+            _loadRunContent('job','run3.md','run3');
+        """)
+        page.wait_for_selector("#run3 .cron-run-primary")
+        assert len(page.locator("#run3 .cron-run-primary").inner_text()) == 600
+        page.evaluate("toggleCronRunExpanded('job','run3.md','run3')")
+        assert len(page.locator("#run3 .cron-run-primary").inner_text()) == 900
+        browser.close()
diff --git a/tests/test_sprint10.py b/tests/test_sprint10.py
index 1f9ddfc9156..4a631f1a519 100644
--- a/tests/test_sprint10.py
+++ b/tests/test_sprint10.py
@@ -181,7 +181,8 @@ def test_cron_history_button_in_panels_js(cleanup_test_sessions):
 
 def test_cron_output_snippet_helper(cleanup_test_sessions):
     src, _ = get_text("/static/panels.js")
-    assert "_cronOutputSnippet" in src
+    assert "data.projection" in src
+    assert "cron-run-primary" in src
 
 
 def test_cron_output_usage_metadata_parses_optional_fields(cleanup_test_sessions):
@@ -223,7 +224,8 @@ def test_cron_output_window_preserves_response_after_large_prompt(cleanup_test_s
     from api.routes import _cron_output_content_window
 
     content = (
-        "Job metadata\n"
+        "# Cron Job: Nightly\n"
+        "**Job ID:** abc123\n"
         "## Prompt\n"
         + ("skill dump\n" * 1200)
         + "user prompt\n"
@@ -236,7 +238,7 @@ def test_cron_output_window_preserves_response_after_large_prompt(cleanup_test_s
     assert len(window) <= 8000
     assert "## Response" in window
     assert "actual useful cron result" in window
-    assert "Job metadata" in window
+    assert "# Cron Job: Nightly" in window
 
 
 def test_cron_output_window_without_response_uses_tail(cleanup_test_sessions):
diff --git a/tests/test_v050257_opus_followups.py b/tests/test_v050257_opus_followups.py
index 5ad392c5a80..21a6940f65b 100644
--- a/tests/test_v050257_opus_followups.py
+++ b/tests/test_v050257_opus_followups.py
@@ -113,6 +113,7 @@ def test_cron_history_rejects_traversal_in_job_id():
         assert 'job_id in (".", "..")' in body, (
             f"{name} must explicitly reject `.` and `..` in addition to the regex."
         )
+    assert "parse_cron_output_artifact" in detail_body
 
 
 # ── 3: int() bounds checking on offset/limit ────────────────────────────────

From ed33c2ecbd6bfb54725918571908e92a8e6a1ad6 Mon Sep 17 00:00:00 2001
From: Rod Boev 
Date: Thu, 27 Aug 2026 10:49:38 -0400
Subject: [PATCH 2/8] fix(#7303): avoid duplicating raw cron payloads

---
 api/routes.py                               | 1 +
 tests/test_issue7303_cron_response_first.py | 2 +-
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/api/routes.py b/api/routes.py
index b8c680fc731..66b47b432c6 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -21761,6 +21761,7 @@ def _handle_cron_run_detail(handler, parsed):
         job = get_job(job_id) if get_job else None
         job_mode = "unknown" if job is None else ("script" if job.get("no_agent") else "agent")
         projection = parse_cron_output_artifact(content, job_mode=job_mode)
+        projection.pop("raw", None)
         snippet = _cron_output_snippet(content, job_mode=job_mode)
         usage = _cron_output_usage_metadata(content, job_mode=job_mode)
         return j(handler, {"job_id": job_id, "filename": filename,
diff --git a/tests/test_issue7303_cron_response_first.py b/tests/test_issue7303_cron_response_first.py
index 4dc32a5eb11..2a8a816eed3 100644
--- a/tests/test_issue7303_cron_response_first.py
+++ b/tests/test_issue7303_cron_response_first.py
@@ -134,7 +134,7 @@ def test_run_detail_returns_shared_projection_and_exact_content(monkeypatch, tmp
     body = json.loads(handler.wfile.getvalue())
     assert handler.status == 200 and body["content"] == raw
     assert body["projection"]["response"] == "literal response"
-    assert body["projection"]["raw"] == raw
+    assert "raw" not in body["projection"]
 
 
 def _load_run_function():

From 3ade5a7363b94380d69567f1ca28165f1f764aad Mon Sep 17 00:00:00 2001
From: Rod Boev 
Date: Fri, 28 Aug 2026 10:57:31 -0400
Subject: [PATCH 3/8] fix(#7303): preserve historical cron run modes

---
 api/cron_output.py                          | 44 ++++++++++
 api/routes.py                               | 31 +++++--
 tests/test_issue7303_cron_response_first.py | 93 ++++++++++++++++++++-
 3 files changed, 160 insertions(+), 8 deletions(-)

diff --git a/api/cron_output.py b/api/cron_output.py
index d8e1a852b7d..be37969e968 100644
--- a/api/cron_output.py
+++ b/api/cron_output.py
@@ -9,6 +9,50 @@
 _PROMPT = re.compile(r"^## Prompt[ \t]*\r?$", re.MULTILINE)
 _HEADING = re.compile(r"^#{1,2} Response[ \t]*\r?$", re.MULTILINE)
 _ERROR = re.compile(r"^#{1,2} Error[ \t]*\r?$", re.MULTILINE)
+_CRON_JOB_LINE = re.compile(r"^# Cron Job:[^\r\n]*\r?$")
+_JOB_ID_LINE = re.compile(r"^\*\*Job ID:\*\*[ \t]*[^\r\n]+\r?$")
+_RUN_TIME_LINE = re.compile(r"^\*\*Run Time:\*\*[ \t]*[^\r\n]+\r?$")
+_SCHEDULE_LINE = re.compile(r"^\*\*Schedule:\*\*[ \t]*[^\r\n]+\r?$")
+_MODE_SCRIPT_LINE = re.compile(r"^\*\*Mode:\*\*[ \t]*no_agent \(script\)[ \t]*\r?$")
+
+
+def _opening_lines(text: str):
+    """Yield only top-level metadata lines before artifact content begins."""
+    for line in text.splitlines(keepends=True):
+        stripped = line.rstrip("\r\n")
+        if stripped == "---" or stripped in ("## Prompt", "## Response", "## Error"):
+            if stripped == "## Prompt":
+                yield stripped
+            break
+        yield stripped
+
+
+def resolve_cron_artifact_mode(text: str, *, legacy_job_mode: str = "unknown") -> str:
+    """Resolve one artifact's mode, using the current job only for legacy files."""
+    raw = text if isinstance(text, str) else str(text or "")
+    lines = list(_opening_lines(raw))
+    if not lines or not _CRON_JOB_LINE.fullmatch(lines[0]):
+        return legacy_job_mode if legacy_job_mode in ("agent", "script") else "unknown"
+
+    positions = {}
+    counts = {"job_id": 0, "run_time": 0, "schedule": 0, "mode": 0}
+    for index, line in enumerate(lines[1:], 1):
+        for name, pattern in (("job_id", _JOB_ID_LINE), ("run_time", _RUN_TIME_LINE),
+                              ("schedule", _SCHEDULE_LINE), ("mode", _MODE_SCRIPT_LINE)):
+            if pattern.fullmatch(line):
+                counts[name] += 1
+                positions.setdefault(name, index)
+                break
+
+    if all(name in positions for name in ("job_id", "run_time", "mode")) and not any(counts[name] > 1 for name in positions):
+        if positions["job_id"] < positions["run_time"] < positions["mode"]:
+            return "script"
+    if all(name in positions for name in ("job_id", "run_time", "schedule")) and not any(counts[name] > 1 for name in positions):
+        prompt = next((index for index, line in enumerate(lines[1:], 1) if _PROMPT.fullmatch(line)), None)
+        if (positions["job_id"] < positions["run_time"] < positions["schedule"] and
+                prompt is not None and positions["schedule"] < prompt):
+            return "agent"
+    return "unknown"
 
 
 def _outside_fence_and_quote(text: str, index: int) -> bool:
diff --git a/api/routes.py b/api/routes.py
index 66b47b432c6..770141b80a8 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -21707,6 +21707,8 @@ def _handle_cron_history(handler, parsed):
     out_dir = CRON_OUT / job_id
     runs = []
     total = 0
+    get_job = getattr(__import__("cron.jobs", fromlist=["get_job"]), "get_job", None)
+    legacy_job_mode = [None]
     if out_dir.exists():
         all_files = sorted(out_dir.glob("*.md"), key=lambda f: f.stat().st_mtime, reverse=True)
         total = len(all_files)
@@ -21715,9 +21717,7 @@ def _handle_cron_history(handler, parsed):
             try:
                 st = f.stat()
                 content = f.read_text(encoding="utf-8", errors="replace")
-                get_job = getattr(__import__("cron.jobs", fromlist=["get_job"]), "get_job", None)
-                job = get_job(job_id) if get_job else None
-                job_mode = "script" if job and job.get("no_agent") else ("agent" if job else "unknown")
+                job_mode = _resolve_cron_artifact_mode(content, get_job, job_id, legacy_job_mode)
                 usage = _cron_output_usage_metadata(content, job_mode=job_mode)
                 runs.append({
                     "filename": f.name,
@@ -21758,8 +21758,7 @@ def _handle_cron_run_detail(handler, parsed):
         from api.cron_output import parse_cron_output_artifact
 
         get_job = getattr(cron_jobs, "get_job", None)
-        job = get_job(job_id) if get_job else None
-        job_mode = "unknown" if job is None else ("script" if job.get("no_agent") else "agent")
+        job_mode = _resolve_cron_artifact_mode(content, get_job, job_id, [None])
         projection = parse_cron_output_artifact(content, job_mode=job_mode)
         projection.pop("raw", None)
         snippet = _cron_output_snippet(content, job_mode=job_mode)
@@ -21835,6 +21834,24 @@ def _floatish(value: str):
     return usage
 
 
+def _cron_legacy_job_mode(get_job, job_id: str) -> str:
+    """Return mutable job mode only as fallback for unclassified legacy artifacts."""
+    job = get_job(job_id) if get_job else None
+    return "script" if job and job.get("no_agent") else ("agent" if job else "unknown")
+
+
+def _resolve_cron_artifact_mode(content, get_job, job_id: str, legacy_mode) -> str:
+    """Classify artifact metadata before consulting mutable legacy job state."""
+    from api.cron_output import resolve_cron_artifact_mode
+
+    mode = resolve_cron_artifact_mode(content)
+    if mode != "unknown":
+        return mode
+    if legacy_mode[0] is None:
+        legacy_mode[0] = _cron_legacy_job_mode(get_job, job_id)
+    return resolve_cron_artifact_mode(content, legacy_job_mode=legacy_mode[0])
+
+
 def _cron_output_snippet(text: str, limit: int = 600, *, job_mode: str = "agent") -> str:
     """Extract the response body from a cron output .md file for preview.
 
@@ -21881,14 +21898,14 @@ def _handle_cron_output(handler, parsed):
         limit = 5
     out_dir = CRON_OUT / job_id
     get_job = getattr(cron_jobs, "get_job", None)
-    job = get_job(job_id) if get_job else None
-    job_mode = "script" if job and job.get("no_agent") else ("agent" if job else "unknown")
+    legacy_job_mode = [None]
     outputs = []
     if out_dir.exists():
         files = sorted(out_dir.glob("*.md"), key=lambda f: f.stat().st_mtime, reverse=True)[:limit]
         for f in files:
             try:
                 txt = f.read_text(encoding="utf-8", errors="replace")
+                job_mode = _resolve_cron_artifact_mode(txt, get_job, job_id, legacy_job_mode)
                 outputs.append({"filename": f.name, "content": _cron_output_content_window(txt, job_mode=job_mode)})
             except Exception:
                 logger.debug("Failed to read cron output file %s", f)
diff --git a/tests/test_issue7303_cron_response_first.py b/tests/test_issue7303_cron_response_first.py
index 2a8a816eed3..ae49ed84823 100644
--- a/tests/test_issue7303_cron_response_first.py
+++ b/tests/test_issue7303_cron_response_first.py
@@ -11,7 +11,16 @@
 
 
 def agent_artifact(response="Done", *, newline="\n"):
-    return newline.join(("# Cron Job: demo", "", "**Job ID:** abc", "", "## Prompt", "", "context", "", "## Response", "", response))
+    return newline.join(("# Cron Job: demo", "**Job ID:** abc", "**Run Time:** 2026-08-27T12:00:00Z", "**Schedule:** daily", "", "## Prompt", "", "context", "", "## Response", "", response))
+
+
+def script_artifact(stdout="script output", *, separator=True):
+    lines = ["# Cron Job: demo", "**Job ID:** abc", "**Run Time:** 2026-08-27T12:00:00Z", "**Mode:** no_agent (script)"]
+    if separator:
+        lines += ["", "---", "", stdout]
+    elif stdout:
+        lines += ["", stdout]
+    return "\n".join(lines)
 
 
 def large_issue_artifact():
@@ -110,6 +119,88 @@ def test_parser_fallbacks_keep_error_and_empty_artifacts_raw():
         assert result["kind"] == "raw" and result["fallback_reason"] == reason
 
 
+def test_resolver_accepts_terminal_script_forms_and_rejects_near_miss():
+    from api.cron_output import resolve_cron_artifact_mode
+
+    assert resolve_cron_artifact_mode(script_artifact("failed", separator=False)) == "script"
+    assert resolve_cron_artifact_mode(script_artifact("", separator=False)) == "script"
+    assert resolve_cron_artifact_mode(agent_artifact(), legacy_job_mode="script") == "agent"
+    near_miss = agent_artifact().replace("**Schedule:** daily\n", "")
+    assert resolve_cron_artifact_mode(near_miss, legacy_job_mode="agent") == "unknown"
+    assert resolve_cron_artifact_mode("legacy output", legacy_job_mode="agent") == "agent"
+    assert resolve_cron_artifact_mode("legacy output", legacy_job_mode="unknown") == "unknown"
+
+
+def test_historical_mode_is_owned_by_each_artifact_across_job_mutations(monkeypatch, tmp_path):
+    import api.routes as routes
+
+    output = tmp_path / "job_abc"
+    output.mkdir()
+    agent = agent_artifact("agent response")
+    script = script_artifact("## Prompt\n\n# Cron Job: fake\n\n## Response\nscript output")
+    (output / "agent.md").write_text(agent, encoding="utf-8")
+    (output / "script.md").write_text(script, encoding="utf-8")
+    jobs = types.ModuleType("cron.jobs")
+    jobs.OUTPUT_DIR = tmp_path
+    current = {"no_agent": True}
+    jobs.get_job = lambda job_id: {"id": job_id, **current}
+    cron = types.ModuleType("cron")
+    cron.__path__ = []
+    monkeypatch.setitem(__import__("sys").modules, "cron", cron)
+    monkeypatch.setitem(__import__("sys").modules, "cron.jobs", jobs)
+
+    def detail(filename):
+        handler = _Handler()
+        routes._handle_cron_run_detail(handler, types.SimpleNamespace(query=f"job_id=job_abc&filename={filename}"))
+        return json.loads(handler.wfile.getvalue())
+
+    script_view = detail("agent.md")
+    assert script_view["projection"]["kind"] == "agent"
+    assert script_view["usage"] == {}
+    current["no_agent"] = False
+    assert detail("script.md")["projection"]["kind"] == "raw"
+
+    history = _Handler()
+    routes._handle_cron_history(history, types.SimpleNamespace(query="job_id=job_abc"))
+    history_body = json.loads(history.wfile.getvalue())
+    assert {run["filename"]: run["usage"] for run in history_body["runs"]} == {"agent.md": {}, "script.md": {}}
+
+    output_handler = _Handler()
+    routes._handle_cron_output(output_handler, types.SimpleNamespace(query="job_id=job_abc"))
+    output_body = json.loads(output_handler.wfile.getvalue())
+    assert {run["filename"]: run["content"] for run in output_body["outputs"]} == {"agent.md": agent, "script.md": script}
+
+
+def test_deleted_job_uses_artifact_mode_but_not_legacy_fallback(monkeypatch, tmp_path):
+    import api.routes as routes
+
+    output = tmp_path / "job_abc"
+    output.mkdir()
+    agent = agent_artifact("answer")
+    legacy = "old output\n## Response\nshould stay raw"
+    (output / "agent.md").write_text(agent, encoding="utf-8")
+    (output / "legacy.md").write_text(legacy, encoding="utf-8")
+    jobs = types.ModuleType("cron.jobs")
+    jobs.OUTPUT_DIR = tmp_path
+    jobs.get_job = lambda job_id: None
+    cron = types.ModuleType("cron")
+    cron.__path__ = []
+    monkeypatch.setitem(__import__("sys").modules, "cron", cron)
+    monkeypatch.setitem(__import__("sys").modules, "cron.jobs", jobs)
+
+    def detail(filename):
+        handler = _Handler()
+        routes._handle_cron_run_detail(handler, types.SimpleNamespace(query=f"job_id=job_abc&filename={filename}"))
+        return json.loads(handler.wfile.getvalue())
+
+    assert detail("agent.md")["projection"]["kind"] == "agent"
+    stored_legacy = legacy.replace("\n", "\r\n")
+    assert detail("legacy.md")["projection"] == {
+        "kind": "raw", "response": None, "diagnostics": stored_legacy,
+        "fallback_reason": "unknown_mode",
+    }
+
+
 class _Handler:
     def __init__(self): self.status, self.wfile = None, io.BytesIO()
     def send_response(self, status): self.status = status

From db4b9aafd1f1afa15576b0fbb11528cd65d5269c Mon Sep 17 00:00:00 2001
From: Rod Boev 
Date: Fri, 28 Aug 2026 11:43:33 -0400
Subject: [PATCH 4/8] fix(#7303): preserve legacy cron artifact modes

---
 api/cron_output.py                          | 45 ++++++++++++++-
 api/routes.py                               | 38 +++++++------
 tests/test_issue7303_cron_response_first.py | 63 +++++++++++++++++----
 3 files changed, 115 insertions(+), 31 deletions(-)

diff --git a/api/cron_output.py b/api/cron_output.py
index be37969e968..46c0d65b073 100644
--- a/api/cron_output.py
+++ b/api/cron_output.py
@@ -14,12 +14,26 @@
 _RUN_TIME_LINE = re.compile(r"^\*\*Run Time:\*\*[ \t]*[^\r\n]+\r?$")
 _SCHEDULE_LINE = re.compile(r"^\*\*Schedule:\*\*[ \t]*[^\r\n]+\r?$")
 _MODE_SCRIPT_LINE = re.compile(r"^\*\*Mode:\*\*[ \t]*no_agent \(script\)[ \t]*\r?$")
+_FENCE_LINE = re.compile(r"^(`{3,}|~{3,})(?:[^`~]*)$")
 
 
 def _opening_lines(text: str):
     """Yield only top-level metadata lines before artifact content begins."""
+    fence_char = None
+    fence_length = 0
     for line in text.splitlines(keepends=True):
         stripped = line.rstrip("\r\n")
+        fence = _FENCE_LINE.fullmatch(stripped.strip())
+        if fence:
+            delimiter = fence.group(1)
+            if fence_char is None:
+                fence_char, fence_length = delimiter[0], len(delimiter)
+            elif delimiter[0] == fence_char and len(delimiter) >= fence_length:
+                fence_char = None
+                fence_length = 0
+            continue
+        if fence_char is not None:
+            continue
         if stripped == "---" or stripped in ("## Prompt", "## Response", "## Error"):
             if stripped == "## Prompt":
                 yield stripped
@@ -27,6 +41,12 @@ def _opening_lines(text: str):
         yield stripped
 
 
+def is_legacy_cron_artifact(text: str) -> bool:
+    """Return whether an artifact has no producer-owned cron envelope."""
+    lines = list(_opening_lines(text if isinstance(text, str) else str(text or "")))
+    return not lines or not _CRON_JOB_LINE.fullmatch(lines[0])
+
+
 def resolve_cron_artifact_mode(text: str, *, legacy_job_mode: str = "unknown") -> str:
     """Resolve one artifact's mode, using the current job only for legacy files."""
     raw = text if isinstance(text, str) else str(text or "")
@@ -73,7 +93,7 @@ def _outside_fence_and_quote(text: str, index: int) -> bool:
     return fence_char is None and not text[line_start:index].lstrip().startswith(">")
 
 
-def parse_cron_output_artifact(text: str, *, job_mode: str = "unknown") -> dict:
+def parse_cron_output_artifact(text: str, *, job_mode: str = "unknown", legacy: bool = False) -> dict:
     """Return one fail-closed, raw-preserving projection for a cron artifact."""
     raw = text if isinstance(text, str) else str(text or "")
     base = {"kind": "raw", "response": None, "diagnostics": raw,
@@ -89,8 +109,27 @@ def parse_cron_output_artifact(text: str, *, job_mode: str = "unknown") -> dict:
                          if _outside_fence_and_quote(raw, m.start())]
     prompt = prompt_candidates[0] if len(prompt_candidates) == 1 else None
     if not preamble or not prompt or prompt.start() <= preamble.end():
-        base["fallback_reason"] = "malformed_preamble"
-        return base
+        if not legacy:
+            base["fallback_reason"] = "malformed_preamble"
+            return base
+        errors = [m for m in _ERROR.finditer(raw)
+                  if _outside_fence_and_quote(raw, m.start())]
+        if errors:
+            base["fallback_reason"] = "error_output"
+            return base
+        candidates = [m for m in _HEADING.finditer(raw)
+                      if _outside_fence_and_quote(raw, m.start())]
+        if len(candidates) != 1:
+            base["fallback_reason"] = "missing_marker" if not candidates else "ambiguous_marker"
+            return base
+        marker = candidates[0]
+        response = raw[marker.end():].lstrip("\r\n")
+        if not response:
+            base["fallback_reason"] = "empty_response"
+            return base
+        return {"kind": "agent", "response": response,
+                "diagnostics": raw[:marker.start()].rstrip(), "raw": raw,
+                "fallback_reason": None}
     errors = [m for m in _ERROR.finditer(raw)
               if _outside_fence_and_quote(raw, m.start())]
     if errors:
diff --git a/api/routes.py b/api/routes.py
index 770141b80a8..a8312497943 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -1420,7 +1420,7 @@ def _is_cron_running(job_id: str) -> tuple[bool, float]:
         return True, time.time() - t
 
 
-def _cron_output_content_window(text: str, limit: int = _CRON_OUTPUT_CONTENT_LIMIT, *, job_mode: str = "agent") -> str:
+def _cron_output_content_window(text: str, limit: int = _CRON_OUTPUT_CONTENT_LIMIT, *, job_mode: str = "agent", legacy: bool = False) -> str:
     """Return a bounded cron output window that preserves useful response text.
 
     Cron output files can contain large skill dumps in the Prompt section. The
@@ -1434,7 +1434,7 @@ def _cron_output_content_window(text: str, limit: int = _CRON_OUTPUT_CONTENT_LIM
 
     from api.cron_output import bounded_cron_projection, parse_cron_output_artifact
 
-    projection = parse_cron_output_artifact(text, job_mode=job_mode)
+    projection = parse_cron_output_artifact(text, job_mode=job_mode, legacy=legacy)
     bounded = bounded_cron_projection(projection, limit)
     if bounded["kind"] == "agent":
         header = bounded["diagnostics"][:_CRON_OUTPUT_HEADER_CONTEXT].rstrip()
@@ -21717,8 +21717,8 @@ def _handle_cron_history(handler, parsed):
             try:
                 st = f.stat()
                 content = f.read_text(encoding="utf-8", errors="replace")
-                job_mode = _resolve_cron_artifact_mode(content, get_job, job_id, legacy_job_mode)
-                usage = _cron_output_usage_metadata(content, job_mode=job_mode)
+                job_mode, legacy = _resolve_cron_artifact_mode(content, get_job, job_id, legacy_job_mode)
+                usage = _cron_output_usage_metadata(content, job_mode=job_mode, legacy=legacy)
                 runs.append({
                     "filename": f.name,
                     "size": st.st_size,
@@ -21758,11 +21758,11 @@ def _handle_cron_run_detail(handler, parsed):
         from api.cron_output import parse_cron_output_artifact
 
         get_job = getattr(cron_jobs, "get_job", None)
-        job_mode = _resolve_cron_artifact_mode(content, get_job, job_id, [None])
-        projection = parse_cron_output_artifact(content, job_mode=job_mode)
+        job_mode, legacy = _resolve_cron_artifact_mode(content, get_job, job_id, [None])
+        projection = parse_cron_output_artifact(content, job_mode=job_mode, legacy=legacy)
         projection.pop("raw", None)
-        snippet = _cron_output_snippet(content, job_mode=job_mode)
-        usage = _cron_output_usage_metadata(content, job_mode=job_mode)
+        snippet = _cron_output_snippet(content, job_mode=job_mode, legacy=legacy)
+        usage = _cron_output_usage_metadata(content, job_mode=job_mode, legacy=legacy)
         return j(handler, {"job_id": job_id, "filename": filename,
                            "content": content, "snippet": snippet,
                            "projection": projection,
@@ -21771,7 +21771,7 @@ def _handle_cron_run_detail(handler, parsed):
         return j(handler, {"error": str(e)}, status=500)
 
 
-def _cron_output_usage_metadata(text: str, *, job_mode: str = "unknown") -> dict:
+def _cron_output_usage_metadata(text: str, *, job_mode: str = "unknown", legacy: bool = False) -> dict:
     """Extract optional token/cost metadata from a cron output markdown file."""
     import re as _re
 
@@ -21779,7 +21779,7 @@ def _cron_output_usage_metadata(text: str, *, job_mode: str = "unknown") -> dict
     if job_mode == "agent":
         from api.cron_output import parse_cron_output_artifact
 
-        projection = parse_cron_output_artifact(text, job_mode=job_mode)
+        projection = parse_cron_output_artifact(text, job_mode=job_mode, legacy=legacy)
         if projection["kind"] == "agent":
             head = projection["diagnostics"]
     usage: dict = {}
@@ -21840,19 +21840,21 @@ def _cron_legacy_job_mode(get_job, job_id: str) -> str:
     return "script" if job and job.get("no_agent") else ("agent" if job else "unknown")
 
 
-def _resolve_cron_artifact_mode(content, get_job, job_id: str, legacy_mode) -> str:
+def _resolve_cron_artifact_mode(content, get_job, job_id: str, legacy_mode) -> tuple[str, bool]:
     """Classify artifact metadata before consulting mutable legacy job state."""
-    from api.cron_output import resolve_cron_artifact_mode
+    from api.cron_output import is_legacy_cron_artifact, resolve_cron_artifact_mode
 
     mode = resolve_cron_artifact_mode(content)
     if mode != "unknown":
-        return mode
+        return mode, False
+    if not is_legacy_cron_artifact(content):
+        return "unknown", False
     if legacy_mode[0] is None:
         legacy_mode[0] = _cron_legacy_job_mode(get_job, job_id)
-    return resolve_cron_artifact_mode(content, legacy_job_mode=legacy_mode[0])
+    return legacy_mode[0], legacy_mode[0] == "agent"
 
 
-def _cron_output_snippet(text: str, limit: int = 600, *, job_mode: str = "agent") -> str:
+def _cron_output_snippet(text: str, limit: int = 600, *, job_mode: str = "agent", legacy: bool = False) -> str:
     """Extract the response body from a cron output .md file for preview.
 
     Contract: cron output files use markdown front-matter followed by a
@@ -21864,7 +21866,7 @@ def _cron_output_snippet(text: str, limit: int = 600, *, job_mode: str = "agent"
     """
     from api.cron_output import bounded_cron_projection, parse_cron_output_artifact
 
-    projection = parse_cron_output_artifact(text, job_mode=job_mode)
+    projection = parse_cron_output_artifact(text, job_mode=job_mode, legacy=legacy)
     projection = bounded_cron_projection(projection, limit)
     body = projection["response"] if projection["kind"] == "agent" else text
     body = (body or "").strip()
@@ -21905,8 +21907,8 @@ def _handle_cron_output(handler, parsed):
         for f in files:
             try:
                 txt = f.read_text(encoding="utf-8", errors="replace")
-                job_mode = _resolve_cron_artifact_mode(txt, get_job, job_id, legacy_job_mode)
-                outputs.append({"filename": f.name, "content": _cron_output_content_window(txt, job_mode=job_mode)})
+                job_mode, legacy = _resolve_cron_artifact_mode(txt, get_job, job_id, legacy_job_mode)
+                outputs.append({"filename": f.name, "content": _cron_output_content_window(txt, job_mode=job_mode, legacy=legacy)})
             except Exception:
                 logger.debug("Failed to read cron output file %s", f)
     return j(handler, {"job_id": job_id, "outputs": outputs})
diff --git a/tests/test_issue7303_cron_response_first.py b/tests/test_issue7303_cron_response_first.py
index ae49ed84823..f76b2552eb8 100644
--- a/tests/test_issue7303_cron_response_first.py
+++ b/tests/test_issue7303_cron_response_first.py
@@ -10,12 +10,18 @@
 import pytest
 
 
-def agent_artifact(response="Done", *, newline="\n"):
-    return newline.join(("# Cron Job: demo", "**Job ID:** abc", "**Run Time:** 2026-08-27T12:00:00Z", "**Schedule:** daily", "", "## Prompt", "", "context", "", "## Response", "", response))
+def agent_artifact(response="Done", *, newline="\n", usage=False):
+    lines = ["# Cron Job: demo", "**Job ID:** abc", "**Run Time:** 2026-08-27T12:00:00Z"]
+    if usage:
+        lines.extend(("**Provider:** agent-provider", "**Model:** agent-model", "**Cost:** $0.12", "**Duration:** 4.5s", "**Tokens:** 100 input, 20 output"))
+    lines.extend(("**Schedule:** daily", "", "## Prompt", "", "context", "", "## Response", "", response))
+    return newline.join(lines)
 
 
-def script_artifact(stdout="script output", *, separator=True):
+def script_artifact(stdout="script output", *, separator=True, status=None):
     lines = ["# Cron Job: demo", "**Job ID:** abc", "**Run Time:** 2026-08-27T12:00:00Z", "**Mode:** no_agent (script)"]
+    if status:
+        lines.append(f"**Status:** {status}")
     if separator:
         lines += ["", "---", "", stdout]
     elif stdout:
@@ -122,11 +128,15 @@ def test_parser_fallbacks_keep_error_and_empty_artifacts_raw():
 def test_resolver_accepts_terminal_script_forms_and_rejects_near_miss():
     from api.cron_output import resolve_cron_artifact_mode
 
-    assert resolve_cron_artifact_mode(script_artifact("failed", separator=False)) == "script"
-    assert resolve_cron_artifact_mode(script_artifact("", separator=False)) == "script"
+    assert resolve_cron_artifact_mode(script_artifact("failed", separator=False, status="script failed")) == "script"
+    assert resolve_cron_artifact_mode(script_artifact("", separator=False, status="silent (empty output)")) == "script"
     assert resolve_cron_artifact_mode(agent_artifact(), legacy_job_mode="script") == "agent"
     near_miss = agent_artifact().replace("**Schedule:** daily\n", "")
     assert resolve_cron_artifact_mode(near_miss, legacy_job_mode="agent") == "unknown"
+    fenced_mode = agent_artifact().replace(
+        "**Schedule:** daily\n", "```md\n**Mode:** no_agent (script)\n```\n**Schedule:** daily\n"
+    )
+    assert resolve_cron_artifact_mode(fenced_mode) == "agent"
     assert resolve_cron_artifact_mode("legacy output", legacy_job_mode="agent") == "agent"
     assert resolve_cron_artifact_mode("legacy output", legacy_job_mode="unknown") == "unknown"
 
@@ -136,8 +146,10 @@ def test_historical_mode_is_owned_by_each_artifact_across_job_mutations(monkeypa
 
     output = tmp_path / "job_abc"
     output.mkdir()
-    agent = agent_artifact("agent response")
-    script = script_artifact("## Prompt\n\n# Cron Job: fake\n\n## Response\nscript output")
+    agent = agent_artifact("agent response", usage=True)
+    script = script_artifact(
+        "## Prompt\n\n# Cron Job: fake\n\n## Response\nscript output\n" + ("script-tail\n" * 1100)
+    )
     (output / "agent.md").write_text(agent, encoding="utf-8")
     (output / "script.md").write_text(script, encoding="utf-8")
     jobs = types.ModuleType("cron.jobs")
@@ -156,19 +168,27 @@ def detail(filename):
 
     script_view = detail("agent.md")
     assert script_view["projection"]["kind"] == "agent"
-    assert script_view["usage"] == {}
+    assert script_view["usage"] == {
+        "provider": "agent-provider", "model": "agent-model", "estimated_cost_usd": 0.12,
+        "duration_seconds": 4.5, "input_tokens": 100, "output_tokens": 20, "total_tokens": 120,
+    }
     current["no_agent"] = False
     assert detail("script.md")["projection"]["kind"] == "raw"
 
     history = _Handler()
     routes._handle_cron_history(history, types.SimpleNamespace(query="job_id=job_abc"))
     history_body = json.loads(history.wfile.getvalue())
-    assert {run["filename"]: run["usage"] for run in history_body["runs"]} == {"agent.md": {}, "script.md": {}}
+    assert history_body["runs"]
+    assert {run["filename"]: run["usage"] for run in history_body["runs"]} == {
+        "agent.md": script_view["usage"], "script.md": {},
+    }
 
     output_handler = _Handler()
     routes._handle_cron_output(output_handler, types.SimpleNamespace(query="job_id=job_abc"))
     output_body = json.loads(output_handler.wfile.getvalue())
-    assert {run["filename"]: run["content"] for run in output_body["outputs"]} == {"agent.md": agent, "script.md": script}
+    assert {run["filename"]: run["content"] for run in output_body["outputs"]} == {
+        "agent.md": agent, "script.md": script[-8000:],
+    }
 
 
 def test_deleted_job_uses_artifact_mode_but_not_legacy_fallback(monkeypatch, tmp_path):
@@ -208,6 +228,29 @@ def send_header(self, *_): pass
     def end_headers(self): pass
 
 
+def test_legacy_agent_fallback_keeps_response_and_excludes_response_usage(monkeypatch, tmp_path):
+    import api.routes as routes
+
+    output = tmp_path / "job_abc"
+    output.mkdir()
+    legacy = "legacy prompt\n## Response\n**Cost:** $999\nanswer"
+    (output / "legacy.md").write_text(legacy, encoding="utf-8")
+    jobs = types.ModuleType("cron.jobs")
+    jobs.OUTPUT_DIR = tmp_path
+    jobs.get_job = lambda job_id: {"id": job_id, "no_agent": False}
+    cron = types.ModuleType("cron")
+    cron.__path__ = []
+    monkeypatch.setitem(__import__("sys").modules, "cron", cron)
+    monkeypatch.setitem(__import__("sys").modules, "cron.jobs", jobs)
+
+    handler = _Handler()
+    routes._handle_cron_run_detail(handler, types.SimpleNamespace(query="job_id=job_abc&filename=legacy.md"))
+    body = json.loads(handler.wfile.getvalue())
+    assert body["projection"]["kind"] == "agent"
+    assert body["snippet"] == "**Cost:** $999\r\nanswer"
+    assert body["usage"] == {}
+
+
 def test_run_detail_returns_shared_projection_and_exact_content(monkeypatch, tmp_path):
     import api.routes as routes
     output = tmp_path / "job_abc" / "run.md"

From f9aeff0ffa4c8138ffd41d47380222afad35eaea Mon Sep 17 00:00:00 2001
From: Rod Boev 
Date: Fri, 28 Aug 2026 11:52:43 -0400
Subject: [PATCH 5/8] fix(#7303): harden cron artifact mode detection

---
 api/cron_output.py                          |  5 +++--
 tests/test_issue7303_cron_response_first.py | 13 ++++++++-----
 2 files changed, 11 insertions(+), 7 deletions(-)

diff --git a/api/cron_output.py b/api/cron_output.py
index 46c0d65b073..0e81305dcbd 100644
--- a/api/cron_output.py
+++ b/api/cron_output.py
@@ -64,8 +64,9 @@ def resolve_cron_artifact_mode(text: str, *, legacy_job_mode: str = "unknown") -
                 positions.setdefault(name, index)
                 break
 
-    if all(name in positions for name in ("job_id", "run_time", "mode")) and not any(counts[name] > 1 for name in positions):
-        if positions["job_id"] < positions["run_time"] < positions["mode"]:
+    if all(name in positions for name in ("job_id", "run_time", "mode")):
+        if (positions["job_id"] < positions["run_time"] < positions["mode"] and
+                positions["mode"] == positions["run_time"] + 1):
             return "script"
     if all(name in positions for name in ("job_id", "run_time", "schedule")) and not any(counts[name] > 1 for name in positions):
         prompt = next((index for index, line in enumerate(lines[1:], 1) if _PROMPT.fullmatch(line)), None)
diff --git a/tests/test_issue7303_cron_response_first.py b/tests/test_issue7303_cron_response_first.py
index f76b2552eb8..077b490b9b6 100644
--- a/tests/test_issue7303_cron_response_first.py
+++ b/tests/test_issue7303_cron_response_first.py
@@ -137,6 +137,10 @@ def test_resolver_accepts_terminal_script_forms_and_rejects_near_miss():
         "**Schedule:** daily\n", "```md\n**Mode:** no_agent (script)\n```\n**Schedule:** daily\n"
     )
     assert resolve_cron_artifact_mode(fenced_mode) == "agent"
+    injected_mode = agent_artifact().replace("**Schedule:** daily\n", "**Schedule:** daily\n**Mode:** no_agent (script)\n")
+    assert resolve_cron_artifact_mode(injected_mode) == "agent"
+    script_metadata = script_artifact("**Job ID:** duplicate\n**Mode:** no_agent (script)", separator=False)
+    assert resolve_cron_artifact_mode(script_metadata) == "script"
     assert resolve_cron_artifact_mode("legacy output", legacy_job_mode="agent") == "agent"
     assert resolve_cron_artifact_mode("legacy output", legacy_job_mode="unknown") == "unknown"
 
@@ -199,7 +203,7 @@ def test_deleted_job_uses_artifact_mode_but_not_legacy_fallback(monkeypatch, tmp
     agent = agent_artifact("answer")
     legacy = "old output\n## Response\nshould stay raw"
     (output / "agent.md").write_text(agent, encoding="utf-8")
-    (output / "legacy.md").write_text(legacy, encoding="utf-8")
+    (output / "legacy.md").write_bytes(legacy.encode("utf-8"))
     jobs = types.ModuleType("cron.jobs")
     jobs.OUTPUT_DIR = tmp_path
     jobs.get_job = lambda job_id: None
@@ -214,9 +218,8 @@ def detail(filename):
         return json.loads(handler.wfile.getvalue())
 
     assert detail("agent.md")["projection"]["kind"] == "agent"
-    stored_legacy = legacy.replace("\n", "\r\n")
     assert detail("legacy.md")["projection"] == {
-        "kind": "raw", "response": None, "diagnostics": stored_legacy,
+        "kind": "raw", "response": None, "diagnostics": legacy,
         "fallback_reason": "unknown_mode",
     }
 
@@ -234,7 +237,7 @@ def test_legacy_agent_fallback_keeps_response_and_excludes_response_usage(monkey
     output = tmp_path / "job_abc"
     output.mkdir()
     legacy = "legacy prompt\n## Response\n**Cost:** $999\nanswer"
-    (output / "legacy.md").write_text(legacy, encoding="utf-8")
+    (output / "legacy.md").write_bytes(legacy.encode("utf-8"))
     jobs = types.ModuleType("cron.jobs")
     jobs.OUTPUT_DIR = tmp_path
     jobs.get_job = lambda job_id: {"id": job_id, "no_agent": False}
@@ -247,7 +250,7 @@ def test_legacy_agent_fallback_keeps_response_and_excludes_response_usage(monkey
     routes._handle_cron_run_detail(handler, types.SimpleNamespace(query="job_id=job_abc&filename=legacy.md"))
     body = json.loads(handler.wfile.getvalue())
     assert body["projection"]["kind"] == "agent"
-    assert body["snippet"] == "**Cost:** $999\r\nanswer"
+    assert body["snippet"] == "**Cost:** $999\nanswer"
     assert body["usage"] == {}
 
 

From dcd3ab40bb81d345a985c8ffaa8f40786b63c854 Mon Sep 17 00:00:00 2001
From: Rod Boev 
Date: Fri, 28 Aug 2026 12:07:56 -0400
Subject: [PATCH 6/8] fix(#7303): confine cron details to their job output

---
 api/routes.py                               | 20 +++++++++--------
 tests/test_issue7303_cron_response_first.py | 25 ++++++++++++++++++++-
 2 files changed, 35 insertions(+), 10 deletions(-)

diff --git a/api/routes.py b/api/routes.py
index a8312497943..6cde2fe00b8 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -21746,9 +21746,10 @@ def _handle_cron_run_detail(handler, parsed):
     # the path resolver).
     if not _re.fullmatch(r"[A-Za-z0-9_-][A-Za-z0-9_.-]{0,63}", job_id) or job_id in (".", ".."):
         return j(handler, {"error": "invalid job_id"}, status=400)
-    # Prevent path traversal — resolve and verify it stays within the job's output dir
-    fpath = (CRON_OUT / job_id / filename).resolve()
-    if not fpath.is_relative_to(CRON_OUT.resolve()):
+    # Prevent path traversal, resolve and verify it stays within this job's directory
+    job_dir = (CRON_OUT / job_id).resolve()
+    fpath = (job_dir / filename).resolve()
+    if not fpath.is_relative_to(job_dir):
         return j(handler, {"error": "invalid filename"}, status=400)
     if not fpath.exists():
         return j(handler, {"error": "run not found"}, status=404)
@@ -21771,17 +21772,18 @@ def _handle_cron_run_detail(handler, parsed):
         return j(handler, {"error": str(e)}, status=500)
 
 
-def _cron_output_usage_metadata(text: str, *, job_mode: str = "unknown", legacy: bool = False) -> dict:
+def _cron_output_usage_metadata(text: str, *, job_mode: str = "agent", legacy: bool = False) -> dict:
     """Extract optional token/cost metadata from a cron output markdown file."""
     import re as _re
 
+    if job_mode != "agent":
+        return {}
     head = text
-    if job_mode == "agent":
-        from api.cron_output import parse_cron_output_artifact
+    from api.cron_output import parse_cron_output_artifact
 
-        projection = parse_cron_output_artifact(text, job_mode=job_mode, legacy=legacy)
-        if projection["kind"] == "agent":
-            head = projection["diagnostics"]
+    projection = parse_cron_output_artifact(text, job_mode=job_mode, legacy=legacy)
+    if projection["kind"] == "agent":
+        head = projection["diagnostics"]
     usage: dict = {}
 
     def _intish(value: str):
diff --git a/tests/test_issue7303_cron_response_first.py b/tests/test_issue7303_cron_response_first.py
index 077b490b9b6..1bdf3ef0e47 100644
--- a/tests/test_issue7303_cron_response_first.py
+++ b/tests/test_issue7303_cron_response_first.py
@@ -152,7 +152,7 @@ def test_historical_mode_is_owned_by_each_artifact_across_job_mutations(monkeypa
     output.mkdir()
     agent = agent_artifact("agent response", usage=True)
     script = script_artifact(
-        "## Prompt\n\n# Cron Job: fake\n\n## Response\nscript output\n" + ("script-tail\n" * 1100)
+        "**Cost:** $999\n## Prompt\n\n# Cron Job: fake\n\n## Response\nscript output\n" + ("script-tail\n" * 1100)
     )
     (output / "agent.md").write_text(agent, encoding="utf-8")
     (output / "script.md").write_text(script, encoding="utf-8")
@@ -224,6 +224,29 @@ def detail(filename):
     }
 
 
+def test_run_detail_rejects_cross_job_filename(monkeypatch, tmp_path):
+    import api.routes as routes
+
+    (tmp_path / "job_abc").mkdir()
+    secret_dir = tmp_path / "job_secret"
+    secret_dir.mkdir()
+    (secret_dir / "secret.md").write_text("SECRET", encoding="utf-8")
+    jobs = types.ModuleType("cron.jobs")
+    jobs.OUTPUT_DIR = tmp_path
+    jobs.get_job = lambda job_id: None
+    cron = types.ModuleType("cron")
+    cron.__path__ = []
+    monkeypatch.setitem(__import__("sys").modules, "cron", cron)
+    monkeypatch.setitem(__import__("sys").modules, "cron.jobs", jobs)
+
+    handler = _Handler()
+    routes._handle_cron_run_detail(
+        handler, types.SimpleNamespace(query="job_id=job_abc&filename=../job_secret/secret.md")
+    )
+    assert handler.status == 400
+    assert json.loads(handler.wfile.getvalue()) == {"error": "invalid filename"}
+
+
 class _Handler:
     def __init__(self): self.status, self.wfile = None, io.BytesIO()
     def send_response(self, status): self.status = status

From 5bcfa04a5e193ff8938becaf2fdd1464cf6f6f7b Mon Sep 17 00:00:00 2001
From: Rod Boev 
Date: Fri, 28 Aug 2026 12:47:40 -0400
Subject: [PATCH 7/8] fix(#7303): isolate cron metadata from output content

---
 api/cron_output.py                          | 28 ++++++++++++++++++++-
 api/routes.py                               |  7 ++----
 tests/test_issue7303_cron_response_first.py |  4 +++
 3 files changed, 33 insertions(+), 6 deletions(-)

diff --git a/api/cron_output.py b/api/cron_output.py
index 0e81305dcbd..8f6a2b7ec47 100644
--- a/api/cron_output.py
+++ b/api/cron_output.py
@@ -15,6 +15,10 @@
 _SCHEDULE_LINE = re.compile(r"^\*\*Schedule:\*\*[ \t]*[^\r\n]+\r?$")
 _MODE_SCRIPT_LINE = re.compile(r"^\*\*Mode:\*\*[ \t]*no_agent \(script\)[ \t]*\r?$")
 _FENCE_LINE = re.compile(r"^(`{3,}|~{3,})(?:[^`~]*)$")
+_PRODUCER_HEADER = re.compile(r"^#\s+(?:Cron|Monitor)\s+Job\b", re.IGNORECASE)
+_USAGE_METADATA_LINE = re.compile(
+    r"^\*\*(?:Provider|Model(?: Used)?|Estimated cost|Cost|Duration|Elapsed|Tokens|Status):\*\*[ \t]*"
+)
 
 
 def _opening_lines(text: str):
@@ -44,7 +48,9 @@ def _opening_lines(text: str):
 def is_legacy_cron_artifact(text: str) -> bool:
     """Return whether an artifact has no producer-owned cron envelope."""
     lines = list(_opening_lines(text if isinstance(text, str) else str(text or "")))
-    return not lines or not _CRON_JOB_LINE.fullmatch(lines[0])
+    return not lines or (
+        not _CRON_JOB_LINE.fullmatch(lines[0]) and not _PRODUCER_HEADER.match(lines[0])
+    )
 
 
 def resolve_cron_artifact_mode(text: str, *, legacy_job_mode: str = "unknown") -> str:
@@ -52,6 +58,8 @@ def resolve_cron_artifact_mode(text: str, *, legacy_job_mode: str = "unknown") -
     raw = text if isinstance(text, str) else str(text or "")
     lines = list(_opening_lines(raw))
     if not lines or not _CRON_JOB_LINE.fullmatch(lines[0]):
+        if lines and _PRODUCER_HEADER.match(lines[0]):
+            return "unknown"
         return legacy_job_mode if legacy_job_mode in ("agent", "script") else "unknown"
 
     positions = {}
@@ -76,6 +84,24 @@ def resolve_cron_artifact_mode(text: str, *, legacy_job_mode: str = "unknown") -
     return "unknown"
 
 
+def cron_artifact_metadata_head(text: str) -> str:
+    """Return only contiguous producer metadata before arbitrary artifact content."""
+    raw = text if isinstance(text, str) else str(text or "")
+    metadata = []
+    for line in _opening_lines(raw):
+        if line in ("## Prompt", "## Response", "# Response", "## Error", "# Error", "---"):
+            break
+        if not line.strip():
+            continue
+        if (_CRON_JOB_LINE.fullmatch(line) or _JOB_ID_LINE.fullmatch(line) or
+                _RUN_TIME_LINE.fullmatch(line) or _SCHEDULE_LINE.fullmatch(line) or
+                _MODE_SCRIPT_LINE.fullmatch(line) or _USAGE_METADATA_LINE.match(line)):
+            metadata.append(line)
+            continue
+        break
+    return "\n".join(metadata)
+
+
 def _outside_fence_and_quote(text: str, index: int) -> bool:
     fence_char = None
     fence_length = 0
diff --git a/api/routes.py b/api/routes.py
index 6cde2fe00b8..e3eb7c7bd64 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -21778,12 +21778,9 @@ def _cron_output_usage_metadata(text: str, *, job_mode: str = "agent", legacy: b
 
     if job_mode != "agent":
         return {}
-    head = text
-    from api.cron_output import parse_cron_output_artifact
+    from api.cron_output import cron_artifact_metadata_head
 
-    projection = parse_cron_output_artifact(text, job_mode=job_mode, legacy=legacy)
-    if projection["kind"] == "agent":
-        head = projection["diagnostics"]
+    head = cron_artifact_metadata_head(text)
     usage: dict = {}
 
     def _intish(value: str):
diff --git a/tests/test_issue7303_cron_response_first.py b/tests/test_issue7303_cron_response_first.py
index 1bdf3ef0e47..f14fb345acb 100644
--- a/tests/test_issue7303_cron_response_first.py
+++ b/tests/test_issue7303_cron_response_first.py
@@ -119,10 +119,13 @@ def test_large_prompt_reproduction_preserves_response_in_bounded_window():
 
 def test_parser_fallbacks_keep_error_and_empty_artifacts_raw():
     from api.cron_output import parse_cron_output_artifact
+    from api.routes import _cron_output_usage_metadata
     for suffix, reason in (("\r\n\r\n## Error\r\n\r\n`oops`", "error_output"), ("\r\n\r\n## Response\r\n", "empty_response")):
         raw = agent_artifact().split("## Response", 1)[0] + suffix
         result = parse_cron_output_artifact(raw, job_mode="agent")
         assert result["kind"] == "raw" and result["fallback_reason"] == reason
+    raw = agent_artifact().replace("## Response", "## Response\n\n**Cost:** $999")
+    assert _cron_output_usage_metadata(raw, job_mode="agent") == {}
 
 
 def test_resolver_accepts_terminal_script_forms_and_rejects_near_miss():
@@ -143,6 +146,7 @@ def test_resolver_accepts_terminal_script_forms_and_rejects_near_miss():
     assert resolve_cron_artifact_mode(script_metadata) == "script"
     assert resolve_cron_artifact_mode("legacy output", legacy_job_mode="agent") == "agent"
     assert resolve_cron_artifact_mode("legacy output", legacy_job_mode="unknown") == "unknown"
+    assert resolve_cron_artifact_mode("# Monitor Job: demo\n## Response\nanswer", legacy_job_mode="agent") == "unknown"
 
 
 def test_historical_mode_is_owned_by_each_artifact_across_job_mutations(monkeypatch, tmp_path):

From 35906b938b07bfc85d8038502f37f74f40799fa2 Mon Sep 17 00:00:00 2001
From: Rod Boev 
Date: Fri, 28 Aug 2026 13:23:30 -0400
Subject: [PATCH 8/8] perf(#7303): scan cron markers in one pass

---
 api/cron_output.py | 70 ++++++++++++++++++++++++++--------------------
 1 file changed, 39 insertions(+), 31 deletions(-)

diff --git a/api/cron_output.py b/api/cron_output.py
index 8f6a2b7ec47..fa36537e9ab 100644
--- a/api/cron_output.py
+++ b/api/cron_output.py
@@ -38,7 +38,7 @@ def _opening_lines(text: str):
             continue
         if fence_char is not None:
             continue
-        if stripped == "---" or stripped in ("## Prompt", "## Response", "## Error"):
+        if stripped == "---" or stripped in ("## Prompt", "## Response", "## Error", "# Response", "# Error"):
             if stripped == "## Prompt":
                 yield stripped
             break
@@ -102,22 +102,34 @@ def cron_artifact_metadata_head(text: str) -> str:
     return "\n".join(metadata)
 
 
-def _outside_fence_and_quote(text: str, index: int) -> bool:
+def _top_level_marker_spans(text: str, start: int = 0) -> dict[str, list[tuple[int, int]]]:
+    """Collect bounded top-level marker spans in one forward scan."""
     fence_char = None
     fence_length = 0
-    for line in text[:index].splitlines():
-        stripped = line.strip()
-        match = re.match(r"(`{3,}|~{3,})(?:[^`~]*)$", stripped)
-        if not match:
+    spans: dict[str, list[tuple[int, int]]] = {"prompt": [], "response": [], "error": []}
+    offset = 0
+    for line in text.splitlines(keepends=True):
+        line_start = offset
+        offset += len(line)
+        stripped = line.rstrip("\n")
+        fence = _FENCE_LINE.fullmatch(stripped.strip())
+        if fence:
+            delimiter = fence.group(1)
+            if fence_char is None:
+                fence_char, fence_length = delimiter[0], len(delimiter)
+            elif delimiter[0] == fence_char and len(delimiter) >= fence_length:
+                fence_char = None
+                fence_length = 0
+            continue
+        if fence_char is not None or line_start < start:
             continue
-        delimiter = match.group(1)
-        if fence_char is None:
-            fence_char, fence_length = delimiter[0], len(delimiter)
-        elif delimiter[0] == fence_char and len(delimiter) >= fence_length:
-            fence_char = None
-            fence_length = 0
-    line_start = text.rfind("\n", 0, index) + 1
-    return fence_char is None and not text[line_start:index].lstrip().startswith(">")
+        for name, pattern in (("prompt", _PROMPT), ("response", _HEADING), ("error", _ERROR)):
+            if len(spans[name]) >= 2:
+                continue
+            match = pattern.fullmatch(stripped)
+            if match and not stripped.lstrip().startswith(">"):
+                spans[name].append((line_start + match.start(), line_start + match.end()))
+    return spans
 
 
 def parse_cron_output_artifact(text: str, *, job_mode: str = "unknown", legacy: bool = False) -> dict:
@@ -132,48 +144,44 @@ def parse_cron_output_artifact(text: str, *, job_mode: str = "unknown", legacy:
         base["fallback_reason"] = "unknown_mode"
         return base
     preamble = _PREAMBLE.match(raw)
-    prompt_candidates = [m for m in _PROMPT.finditer(raw, preamble.end() if preamble else 0)
-                         if _outside_fence_and_quote(raw, m.start())]
+    marker_spans = _top_level_marker_spans(raw, preamble.end() if preamble else 0)
+    prompt_candidates = marker_spans["prompt"]
     prompt = prompt_candidates[0] if len(prompt_candidates) == 1 else None
-    if not preamble or not prompt or prompt.start() <= preamble.end():
+    if not preamble or not prompt or prompt[0] <= preamble.end():
         if not legacy:
             base["fallback_reason"] = "malformed_preamble"
             return base
-        errors = [m for m in _ERROR.finditer(raw)
-                  if _outside_fence_and_quote(raw, m.start())]
+        errors = marker_spans["error"]
         if errors:
             base["fallback_reason"] = "error_output"
             return base
-        candidates = [m for m in _HEADING.finditer(raw)
-                      if _outside_fence_and_quote(raw, m.start())]
+        candidates = marker_spans["response"]
         if len(candidates) != 1:
             base["fallback_reason"] = "missing_marker" if not candidates else "ambiguous_marker"
             return base
-        marker = candidates[0]
-        response = raw[marker.end():].lstrip("\r\n")
+        marker_start, marker_end = candidates[0]
+        response = raw[marker_end:].lstrip("\r\n")
         if not response:
             base["fallback_reason"] = "empty_response"
             return base
         return {"kind": "agent", "response": response,
-                "diagnostics": raw[:marker.start()].rstrip(), "raw": raw,
+                "diagnostics": raw[:marker_start].rstrip(), "raw": raw,
                 "fallback_reason": None}
-    errors = [m for m in _ERROR.finditer(raw)
-              if _outside_fence_and_quote(raw, m.start())]
+    errors = marker_spans["error"]
     if errors:
         base["fallback_reason"] = "error_output"
         return base
-    candidates = [m for m in _HEADING.finditer(raw, prompt.end())
-                  if _outside_fence_and_quote(raw, m.start())]
+    candidates = [span for span in marker_spans["response"] if span[0] >= prompt[1]]
     if len(candidates) != 1:
         base["fallback_reason"] = "missing_marker" if not candidates else "ambiguous_marker"
         return base
-    marker = candidates[0]
-    response = raw[marker.end():].lstrip("\r\n")
+    marker_start, marker_end = candidates[0]
+    response = raw[marker_end:].lstrip("\r\n")
     if not response:
         base["fallback_reason"] = "empty_response"
         return base
     return {"kind": "agent", "response": response,
-            "diagnostics": raw[:marker.start()].rstrip(), "raw": raw,
+            "diagnostics": raw[:marker_start].rstrip(), "raw": raw,
             "fallback_reason": None}