diff --git "a/app_kimi.py" "b/app_kimi.py" --- "a/app_kimi.py" +++ "b/app_kimi.py" @@ -1,14 +1,90 @@ """ gijl/agentic-thinker-v1 -- multi-loop dataset generator (single-file app.py) -v8, worker "kimi" (Kimi-Dev-72B): llama.cpp backend + multi-loop + verified, source-grouped, auditable - output, + genuine agent-task trajectories + five specialized workers (glm/minimax/deepseek/kimi/ - qwen) concurrent collection, now with real per-model output-compatibility handling. Sibling files - run the identical pipeline -- see WHAT CHANGED IN v7 and WHAT CHANGED IN v8. +v10, worker "kimi" (Kimi-Dev-72B): llama.cpp backend + multi-loop + verified, auditable output, a + code-enforced general agent-thinker trajectory state machine (think/act/propose/verify/error- + recovery/review), single-file-per-worker storage merged into one public file, and five + specialized, anonymized workers with real per-model output-compatibility handling. Sibling files + run the identical pipeline -- see WHAT CHANGED IN v7, WHAT CHANGED IN v8 and WHAT CHANGED IN v10. SCOPE: this collects data only. It does NOT train anything -- no DPO loop, no optimizer, nothing that touches gradients. Output is training-ready JSONL for someone else's training scripts to consume. +WHAT CHANGED IN v10 (code-enforced general agent-thinker trajectories, single-file-per-worker storage, +anonymization, Dataset Viewer fix) +---------------------------------------------------------------------------------------------------------------- +Trigger: a direct review of the live agent_trajectory data found trajectories too shallow to reflect genuine +"agent-thinker" behavior -- shallow tool use, no real verification step, no error-recovery behavior -- traced to +run_agent_task() being one big free-running system prompt that only SUGGESTED a think/verify/review discipline +in prose, never enforced it. Requirement: the harness's own Python control flow, not the model's prompted +self-discipline (the five teacher models were never trained to self-impose this), must drive that discipline -- +but as a GENERAL framework usable across arbitrarily diverse task kinds, not one fixed scripted routine (a +hard-coded "always propose code, always web-search the error" procedure would leave a model trained on it able +to execute only that one routine). Everything below is shared-pipeline, affects all 5 workers identically unless +marked otherwise, and is this project's one-time, explicit authorization to rewrite run_agent_task()/storage/ +schema -- previously off-limits under the standing "touch only per-worker model identity" rule. + * run_agent_task() REWRITTEN as an explicit, harness-driven phase machine: THINK (restate the task + plan, + decide if tool use is needed -- no tool calls yet) -> ACT (the existing bounded ReAct tool loop, reused + unchanged; zero tool calls is a hard failure unless THINK judged nothing checkable was involved) -> PROPOSE + (draft the candidate answer) -> VERIFY, chosen by the HARNESS from what PROPOSE actually produced, not + hard-coded per task: a fenced code block -> real code execution (new execute_code_for_verification(), + harness-internal only, never a model-callable tool -- best-effort subprocess sandbox: fresh tempdir, + resource.setrlimit for CPU/memory/core-dumps, a hard wall-clock timeout, explicitly NOT a hard security + boundary, acceptable since the code is this pipeline's own generated content); concrete claims grounded in + gathered tool results -> a deterministic grounding check; neither -> a lightweight self-consistency model + call, still logged as a real verify step so "verify what you produce" is visible even off the code/research + path. On failure -> ERROR_RECOVERY: real tool calls search external sources for the exact error or an + already-solved instance of the same problem, then one call reconciles whatever was found into a single + consolidated fix (never patches in every suggestion blindly) -- loops back to VERIFY, bounded by + AGENT_TASK_MAX_VERIFY_ATTEMPTS (default 3). Exhausting attempts is kept as outcome="verify_exhausted" + (trainable_for=["trajectory_discipline"] only, not discarded -- an honest "I could not verify this" is + itself a worth-training example) rather than silently dropped. Then REVIEW (one bounded final pass, at most + one small revision) -> OUTPUT. _VERIFY_BOOLS/_VERIFY_TEMPLATES["agent_task"] extended to match ( + "verified_before_claiming" split into "real_verification_used"/"honest_about_limits", now checkable against + the actual recorded attempts instead of inferred). + * UNIFIED RECORD SCHEMA: every data_type now shares one flat superset envelope (ALL_RECORD_FIELDS = + _SHARED_ENVELOPE_FIELDS + _TYPE_SPECIFIC_FIELDS) instead of each data_type's row carrying only its own + fields -- _normalize_envelope_fields(), now called from the single append_record() write path, fills every + missing field with None AND strips any field outside this set (a malformed/legacy-shaped row cannot + reintroduce column drift). agent_trajectory's shape is now task_prompt/think/research_steps/attempts (each + with its own verification + optional error_recovery)/review/output, replacing the old flat steps/ + final_answer pair. SCHEMA_VERSION bumped 3->4. + * STORAGE CONSOLIDATED to one physical file per worker instead of 7 data_type folders x 3 splits (21 files): + SHARD_PATH = data/_shards/{shard_id}.jsonl, where shard_id is a deterministic sha256 of MODEL_TAG (not the + literal tag -- the Hub's file listing must not reveal which model produced which shard, see anonymization + below). `split` is now a row column, not a folder/filename choice. Still one writer per physical file (every + worker only ever appends to its own shard), so the existing flock-based atomic append stays exactly as + race-free as the old per-type-folder layout. A derived, single public file (MERGED_REPO_PATH = + data/dataset.jsonl) is rebuilt FRESH on every sync by merge_shards_for_sync(): downloads every worker's + current shard, de-duplicates by id, sorts deterministically -- pure and idempotent, so two workers' merges + racing on the hub leaves the loser at worst momentarily missing a few rows, never corrupted, self-healing at + the next sync. DedupIndex/write_status/build_manifest/build_audit all rewritten to one pass over a worker's + own shard instead of the old DATA_TYPES x SPLITS nested-file iteration, same tallies either way. + * DATASET VIEWER FIX: root cause was exactly the "mixed key sets inside one JSONL break the Hub's dataset + viewer" hazard the old _backfill() comment already flagged, now impossible by construction (see unified + schema above) rather than patched after the fact. build_readme()'s YAML now declares ONE config pointing at + data/dataset.jsonl with one physical split name ("train"); the in-row `split` column is the real partition, + documented in the README body. + * ANONYMIZATION: `generated_by`/`collector_tag` fields removed from the schema entirely (not blanked -- absent + from every record going forward); build_readme()'s generated text no longer mentions MODEL_ID/MODEL_TAG or + any model name anywhere. kimi's cross_review rows no longer store which worker wrote the row being reviewed + (reviewed_id, a sha256 hash, already uniquely identifies it on its own) and its DATA_TYPES entry/README text + no longer names Kimi-Dev-72B. MODEL_TAG's existing LOCAL/internal use (STATE_DIR, dedup index, shard_id + input) is explicitly exempt and unchanged -- only public-facing record fields and the README card are + affected. + * KIMI-ONLY: an additive, stricter Arabic-purity check (field evidence: live kimi output showed stray CJK/ + Latin fragments mixed into otherwise-Arabic text that the shared majority-script check, ar/letters >= 0.5, + unchanged for the other 4 workers, does not catch). Wired via the existing additive _EXTRA_LANGUAGE_CHECK + hook in _check_language_match() -- consulted only AFTER the shared check already passed, so this can only + make kimi's own Arabic check stricter, never looser, and is a no-op for the other 4 workers. + * minimax's existing agent_task specialization nudge ("treat it as part of a bigger build") re-anchored from + the now-removed single AGENT_TASK_SYSTEM_TEMPLATE onto the new ACT_SYSTEM_TEMPLATE -- same intent (investigate + related files/tests/config, not just the one symptom), same worker-only scope. + * NOTHING ELSE CHANGED: the scouting/run_cycle control flow, the tool registry, the atem: protocol, llama.cpp + backend/per-model client classes (LlamaCppClient/MinimaxThinkBudgetClient), judgment_dpo/sft_helpful/ + reasoning_qa generation logic, and the v7/v8 per-model output-compatibility mechanisms above are all + untouched by this pass. + WHAT CHANGED IN v8 (deep rethink of cross-worker output COMPATIBILITY, not just schema) ---------------------------------------------------------------------------------------- Trigger: having five workers write byte-identical SCHEMA (v7) does not mean their raw text is equivalent -- each @@ -300,9 +376,10 @@ differently -- see WHAT CHANGED IN v7 and WHAT CHANGED IN v8 at the top of this evidence behind each choice. all five write into the SAME dataset concurrently -- see MODEL_TAG and the v6 note above for how that stays race-free. The chat template + tokenizer come from the MODEL_ID repo (rendered locally), the weights come from the GGUF repo. ONE llama-server process, ONE generation stream (parallel slots = 1), context 12288 tokens by default (GIJL_CTX). The tool-call XML convention below -() is taught IN THE PROMPT (see build_scout_prompt/build_agent_task_prompt), not a trained-in format of -any one model, so it is unchanged across workers -- but it is still worth a GIJL_SMOKE=1 run per worker before a -long session, to confirm this model actually follows it turn after turn. +() is taught IN THE PROMPT (see build_scout_prompt and the THINK/ACT/PROPOSE/RECOVERY_SEARCH phase templates +in the v10 agent_task state machine), not a trained-in format of any one model, so it is unchanged across workers -- +but it is still worth a GIJL_SMOKE=1 run per worker before a long session, to confirm this model actually +follows it turn after turn. HARDWARE (llama.cpp backend): NVIDIA GPU via a CUDA build, otherwise CPU. llama-server's --fit (default) fits layers to the free VRAM; when the whole model fits, every layer is offloaded (-ngl 999). TPU is not supported by llama.cpp. @@ -328,9 +405,13 @@ import time, so the module can be imported/tested without side effects; set GIJL_SKIP_INSTALL=1 to skip them. RECORD SCHEMAS: see the auto-generated README.md / MANIFEST.json, or -DATA_TYPES below. Shared envelope on every record: id, schema_version, -data_type, loop, language, content_hash, source_kind, source_url, -generated_by, generated_at. +DATA_TYPES below. Shared envelope on every record (v10): id, schema_version, +data_type, loop, language, split, content_hash, source_kind, source_url, +generated_at, run_id, focus, source_license, source_chars, verified, +verifier_score, trainable_for -- see ALL_RECORD_FIELDS / _normalize_envelope_fields() +for the full column set every row carries regardless of data_type (no +generated_by/collector_tag anymore -- v10 anonymized the public schema, see the +v10 changelog below and the note at append_record()). PRIVACY / LICENSE NOTE: scraped third-party text is NOT republished by default -- source_index stores metadata (url, title, length, sha256) only. @@ -374,6 +455,7 @@ import shutil import socket import subprocess import tarfile +import tempfile import threading import traceback import urllib.robotparser @@ -452,15 +534,20 @@ WORK_DIR = _default_work_dir() # reads or writes is namespaced under MODEL_TAG. This is the actual fix; create_repo()/sync_to_hf()'s retry # (below) is only defense in depth for the hub's own commit-head race on top of that. MODEL_TAG = os.environ.get("GIJL_MODEL_TAG", "kimi") # "glm" | "minimax" | "deepseek" | "kimi" | "qwen" -- set per worker file/Space -STATE_DIR = os.path.join(WORK_DIR, "state", MODEL_TAG) +STATE_DIR = os.path.join(WORK_DIR, "state", MODEL_TAG) # internal/local only -- exempt from v10 anonymization STATUS_PATH = os.path.join(STATE_DIR, "status.json") SEEN_PATH = os.path.join(STATE_DIR, "seen.json") -MANIFEST_PATH = os.path.join(WORK_DIR, f"MANIFEST.{MODEL_TAG}.json") -README_PATH = os.path.join(WORK_DIR, f"README.{MODEL_TAG}.md") +# v10: un-namespaced (no more MODEL_TAG suffix) -- these are now the GLOBAL manifest/README, built from the +# cross-worker merge (see merge_shards_for_sync()/sync_to_hf()) rather than one description per worker shard. +MANIFEST_PATH = os.path.join(WORK_DIR, "MANIFEST.json") +README_PATH = os.path.join(WORK_DIR, "README.md") SPLITS = ("train", "validation", "test") SPLIT_WEIGHTS = {"train": 90, "validation": 5, "test": 5} # deterministic, hash-based -- see split_for_id() -SCHEMA_VERSION = 3 +# v10: bumped 3->4 -- the superset envelope (ALL_RECORD_FIELDS/_normalize_envelope_fields), the removal of +# generated_by/collector_tag, and the new agent_trajectory shape (think/research_steps/attempts/review/output) +# are all real row-shape changes, not just new optional fields. +SCHEMA_VERSION = 4 DATA_TYPES = { "preference_pair": { @@ -496,31 +583,49 @@ DATA_TYPES = { "folder": "source_index", "title": "Source index (provenance)", "description": "one row per scenario tried: url, kind, title, length, sha256, status (used/rejected), " "error. Bodies are NOT stored unless GIJL_STORE_SOURCE_BODIES=1."}, - # v7, kimi-only (see WHAT CHANGED IN v7): Kimi's second opinion on another worker's borderline - # agent_trajectory/sft_chat row. Never written by any other worker; purely additive, joins on - # reviewed_id. Does not exist in glm/minimax/deepseek/qwen's DATA_TYPES. + # v10, this worker only (see WHAT CHANGED IN v10): a second opinion on another worker's borderline + # agent_trajectory/sft_chat row. Never written by any other worker; purely additive, joins on reviewed_id + # (globally unique, see _rid()) -- no worker identity is stored in the row itself, per this project's + # anonymization requirement. Does not exist in the other four workers' DATA_TYPES. "cross_review": { - "folder": "cross_review_dataset", "title": "Cross-worker review (kimi, v7)", - "description": "Kimi-Dev-72B's independent re-verification of another worker's agent_trajectory/sft_chat " - "row that was not verified=='pass' on first pass. reviewed_tag/reviewed_id/reviewed_type " - "identify the row being reviewed; verdict/score/issues are Kimi's own opinion, in the same " - "shape as that loop's normal verifier output. This is a second, differently-trained model's " - "judgment -- the self-review limitation noted at _VERIFY_HEAD does not apply to it."}, + "folder": "cross_review_dataset", "title": "Cross-worker review", + "description": "An independent re-verification of another worker's agent_trajectory/sft_chat row that " + "was not verified=='pass' on first pass, by a differently-trained model than the one that " + "wrote it. reviewed_id identifies the row being reviewed; verdict/score/issues are the " + "reviewer's own opinion, in the same shape as that loop's normal verifier output. This is " + "a second model's judgment -- the self-review limitation noted at _VERIFY_HEAD does not " + "apply to it."}, } -DATASET_DIR = os.path.join(WORK_DIR, DATA_TYPES["preference_pair"]["folder"], MODEL_TAG) # legacy alias (old contract) - - -def type_dir(data_type: str) -> str: - # v6: nested under MODEL_TAG -- e.g. dpo_dataset/glm/train.jsonl, never dpo_dataset/train.jsonl directly -- so - # the glm/minimax/deepseek workers' upload_folder() calls touch disjoint sets of paths. fnmatch (used by - # huggingface_hub's allow_patterns) already matches "/" with "*", so _sync_patterns() needed no change. - return os.path.join(WORK_DIR, DATA_TYPES[data_type]["folder"], MODEL_TAG) +def _shard_id_for(tag: str) -> str: + # v10: ONE physical file per worker instead of 7 data_type folders x 3 splits (21 files/worker) -- see the v10 + # storage note below. Deterministic from MODEL_TAG so no coordination between workers is needed, but it does + # NOT reveal the model name/tag in the Hub's "Files and versions" listing -- the user asked that nothing + # public (records, README, file names) name which worker produced a row; a literal "glm.jsonl"/"minimax.jsonl" + # path would leak exactly that even though the records themselves carry no model-identifying field anymore + # (see _envelope() / build_readme()). + return hashlib.sha256(f"gijl-worker-{tag}".encode("utf-8")).hexdigest()[:10] + + +SHARD_ID = _shard_id_for(MODEL_TAG) +SHARD_DIR = os.path.join(WORK_DIR, "data", "_shards") +SHARD_PATH = os.path.join(SHARD_DIR, f"{SHARD_ID}.jsonl") # THIS worker's own file: single writer, flock-appended, + # same race-freedom MODEL_TAG namespacing gave the old + # per-type folders -- see append_record() below. +MERGED_REPO_PATH = "data/dataset.jsonl" # path as it appears in the Hub repo (README configs: + # block below and the upload call both use this) +MERGED_PATH = os.path.join(WORK_DIR, "data", "dataset.jsonl") # derived: every worker's shard concatenated. Rebuilt + # fresh from the shards on every sync (see + # merge_shards_for_sync()) -- never written to + # directly, so it is safe for >1 worker to + # (re)generate and upload it: worst case under a + # concurrent sync is a few rows momentarily missing, + # which the next sync (every SYNC_INTERVAL_S) fixes. +DATASET_DIR = SHARD_DIR # legacy alias (old contract; a few older comments still refer to this name) def ensure_layout(): os.makedirs(STATE_DIR, exist_ok=True) - for dt in DATA_TYPES: - os.makedirs(type_dir(dt), exist_ok=True) + os.makedirs(SHARD_DIR, exist_ok=True) # v6, worker "glm": zai-org/GLM-4.5-Air -- 106B total/12B active MoE, MIT, fits the 120GB RAM budget at a high @@ -1017,6 +1122,28 @@ _INLINE_CODE_RE = re.compile(r"`[^`\n]*`") _URL_RE = re.compile(r"https?://\S+") +# v10: worker-specific stricter language check hook. None (default) is a no-op for every worker except +# the one build_workers.py wires it for -- currently only kimi (see KIMI_STRICT_LANGUAGE_CHECK there): +# live kimi output showed stray CJK/Latin fragments (Chinese/Korean/garbled English words) mixed into +# otherwise-Arabic text, which the majority-script check below (ar/letters >= 0.5) does not catch, +# since a handful of foreign-script words rarely drags Arabic's share below 50%. Deliberately NOT a +# change to the shared check itself -- the other 4 workers have not shown this problem -- so this is an +# additive hook, consulted only after the existing check already passed. +def _kimi_strict_language_check(language, ar, la, cc, letters): + """v10, this worker only: caps non-Arabic-script (CJK/Cyrillic) contamination on the Arabic branch at 3% + of all letters -- much tighter than the English branch's own 30% CJK cap a few lines up, because a + handful of stray CJK/Latin fragments is exactly the live-dataset contamination this hook exists to catch, + not legitimate quoting (Arabic prose legitimately quoting English code/terms is already handled upstream: + code fences, inline code and URLs are stripped before any of these ratios are computed). A non-Arabic + target language is untouched -- returns True unconditionally so this never changes behavior there.""" + if language != "arabic": + return True + return cc / letters <= 0.03 + + +_EXTRA_LANGUAGE_CHECK = _kimi_strict_language_check + + def _check_language_match(text: str, language: str) -> bool: """FIX: the old Arabic test only failed when there were <20 Arabic letters, so an English answer with 27 Arabic letters passed. Now judged on the RATIO @@ -1031,8 +1158,12 @@ def _check_language_match(text: str, language: str) -> bool: if letters < 8: return "```" in text # pure code answer: nothing to judge if language == "arabic": - return ar / letters >= 0.5 - return (ar / letters <= 0.15) and (cc / letters <= 0.3) + ok = ar / letters >= 0.5 + else: + ok = (ar / letters <= 0.15) and (cc / letters <= 0.3) + if ok and _EXTRA_LANGUAGE_CHECK is not None: + ok = _EXTRA_LANGUAGE_CHECK(language, ar, la, cc, letters) + return ok # --- safety filter for `rejected` ------------------------------------------ @@ -1565,15 +1696,27 @@ def download_llama_cpu_release(): arch = "arm64" if platform.machine().lower() in ("aarch64", "arm64") else "x64" base = "https://github.com/ggml-org/llama.cpp/releases" hdr = {"User-Agent": _CRAWLER_UA} - tags = [] - try: - latest = requests.get(f"{base}/latest", timeout=30, headers=hdr).url.rstrip("/").rsplit("/", 1)[-1] - nt = requests.get(f"{base}/download/{latest}/nightly-tag.txt", timeout=30, headers=hdr) - if nt.status_code == 200 and nt.text.strip(): - tags.append(nt.text.strip().split()[0]) - tags.append(latest) - except requests.RequestException as e: - raise RuntimeError(f"cannot reach GitHub releases: {_redact(str(e))[:200]}") + pinned_tag = os.environ.get("GIJL_LLAMA_CPU_TAG", "").strip() + if pinned_tag: + # v8: bisecting knob, opt-in only (default behavior below is unchanged) -- added because DeepSeek-V4-Flash + # crashed llama-server on CPU with GGML_ASSERT(*cur_backend_id != -1) in ggml_backend_sched_split_graph + # during context/graph setup, a signature consistent with "this architecture's CPU backend coverage in + # llama.cpp is still incomplete" rather than anything wrong in this file -- DeepSeek-V4 support was still + # WIP/fork-based in llama.cpp as of its own GitHub support thread. "latest" always means whatever today's + # CI build is, so there is no way to test a different one without this. EXCLUSIVE on purpose (no silent + # fallback to latest): the point is to know for certain which build actually ran. + print(f"[llama] GIJL_LLAMA_CPU_TAG={pinned_tag} set -- using this exact release instead of latest", flush=True) + tags = [pinned_tag] + else: + tags = [] + try: + latest = requests.get(f"{base}/latest", timeout=30, headers=hdr).url.rstrip("/").rsplit("/", 1)[-1] + nt = requests.get(f"{base}/download/{latest}/nightly-tag.txt", timeout=30, headers=hdr) + if nt.status_code == 200 and nt.text.strip(): + tags.append(nt.text.strip().split()[0]) + tags.append(latest) + except requests.RequestException as e: + raise RuntimeError(f"cannot reach GitHub releases: {_redact(str(e))[:200]}") last_err = "no candidate tag" for tag in tags: dest = os.path.join(LLAMA_CPU_DIR, tag) @@ -1928,7 +2071,7 @@ def start_llama_backend(): # --------------------------------------------------------------------------- # 6. Harness tool-call protocol ("Onyx ATEM": tool calls as XML). This is TAUGHT IN THE -# SYSTEM PROMPT with a worked example (see build_scout_prompt/build_agent_task_prompt), not a trained-in +# SYSTEM PROMPT with a worked example (see build_scout_prompt and the agent_task phase prompts), not a trained-in # format of any one model -- so it is unchanged across the glm/minimax/deepseek workers (v6). The `to=self` / # `to=user` channel-noise stripper below was written for Muse-Glimmer's OWN vocabulary quirk specifically; it # is a harmless no-op for the other two models (nothing to strip), left as-is rather than removed. @@ -2895,15 +3038,17 @@ def _normalize_label(raw, allowed: dict): def _envelope(data_type: str, loop_name: str, rid: str, language: str, scenario: dict, content_text: str, plan=None) -> dict: + # v10: no generated_by/collector_tag -- see the anonymization note at append_record(). `split` is filled + # in later by append_record() (it needs the final record, not just the envelope, to hash consistently). url = scenario["url"] return {"id": rid, "schema_version": SCHEMA_VERSION, "data_type": data_type, "loop": loop_name, "language": language, "content_hash": _content_hash(content_text), "source_kind": scenario["kind"], "source_url": url, - "generated_by": MODEL_ID, "generated_at": _now(), + "generated_at": _now(), "run_id": RUN_ID, "focus": plan.focus_label if plan else "", "source_license": _license_hint(url, scenario["kind"]), "source_chars": len(scenario.get("body") or ""), - "verified": "unverified", "verifier_score": 0, "collector_tag": MODEL_TAG} + "verified": "unverified", "verifier_score": 0, "trainable_for": None} def _rid(*parts) -> str: @@ -3207,32 +3352,163 @@ def generate_records(client, plan: CyclePlan, scenario: dict) -> dict: # --------------------------------------------------------------------------- -# 10a. Agent-task generation (v6). The model solves a REAL (simulated) task with the SAME tool registry as -# scouting -- not "find a scenario for the dataset" -- so the resulting trajectory is genuine agent -# behavior, not a record of how the harness went shopping for raw material. Budget is explicit and separate -# from the scouting cycle's own budget (Anthropic, "Building effective agents": most agent tasks need 5-10 -# tool calls; an unbounded loop is not "thinking harder", it is stalling). Verified as a WHOLE trajectory -# against named principles (_VERIFY_TEMPLATES["agent_task"] below), not a single final-text score. +# 10a. Agent-task generation (v10): a CODE-ENFORCED state machine, not one free-running ReAct loop with +# think/verify/review merely suggested in a system prompt. THINK -> ACT -> PROPOSE -> VERIFY -> +# (ERROR_RECOVERY -> PROPOSE -> VERIFY)* -> REVIEW -> OUTPUT. The harness decides what phase comes +# next and sends a narrow, phase-specific prompt each time; the model only ever fills in that phase's +# content. This replaces the v6 design (one big system prompt + a single post-hoc LLM-judge call, +# with no retry) because the five teacher models were never trained to impose this discipline on +# themselves and did it unreliably when only asked to in a prompt -- the actual redesign requirement. +# +# This is a GENERALIZATION, not one fixed routine: which VERIFY method fires (code_execution / +# grounding / self_consistency) is decided by the harness from what PROPOSE actually produced, never +# hard-coded to "always code" or "always search a bug tracker" -- a model trained on one fixed script +# would be unable to do anything else. The scenario-mining stage above already feeds this a wide +# diversity of real material (Stack Exchange, GitHub issues, OSV advisories, HN, arXiv, Wikipedia, +# arbitrary pages); this phase machine is the constant *shape* applied across all of them, not a +# narrowing to one of them. Verified again, as a WHOLE trajectory, by the post-hoc LLM judge below +# (_VERIFY_TEMPLATES["agent_task"]) -- that pass stays as a second, independent check; it does not +# replace the real, harness-run verification that now happens inside this function. # --------------------------------------------------------------------------- AGENT_TASK_MAX_TOOL_CALLS = int(os.environ.get("GIJL_AGENT_TASK_MAX_CALLS", "8")) AGENT_TASK_MAX_NUDGES = 2 +AGENT_TASK_MAX_VERIFY_ATTEMPTS = int(os.environ.get("GIJL_AGENT_TASK_MAX_VERIFY_ATTEMPTS", "3")) +AGENT_TASK_MAX_RECOVERY_SEARCHES = int(os.environ.get("GIJL_AGENT_TASK_MAX_RECOVERY_SEARCHES", "3")) +THINK_MAX_NEW_TOKENS = 500 +PROPOSE_MAX_NEW_TOKENS = GEN_MAX_NEW_TOKENS +REVIEW_MAX_NEW_TOKENS = 400 _TASK_TAG_RE = re.compile(r"(.*?)", re.DOTALL | re.IGNORECASE) +_NEEDS_TOOLS_RE = re.compile(r"NEEDS_TOOLS\s*:\s*(yes|no)", re.IGNORECASE) +_REVISION_RE = re.compile(r"REVISION\s*:\s*(.*)", re.DOTALL | re.IGNORECASE) +_RATIONALE_RE = re.compile(r"RATIONALE:\s*(.*)") _PY_CODE_BLOCK_RE = re.compile(r"```(?:python|py)?\s*\n(.*?)```", re.DOTALL) -AGENT_TASK_FINISH_SCHEMA = _fn( - "finish_task_answer", - "Give your FINAL answer to the person's task now. Call this once you are done investigating -- not before " - "you have enough to answer well, and not after (do not keep 'double-checking' past the point of usefulness).", - {"final_answer": _S, "confidence": {"type": "string", "description": "low|medium|high"}}, - ["final_answer"]) +EXECUTE_CODE_TIMEOUT_S = int(os.environ.get("GIJL_EXECUTE_CODE_TIMEOUT_S", "20")) +EXECUTE_CODE_MAX_OUTPUT_CHARS = 4000 + + +def _limit_resources(): + """preexec_fn for the code-verification subprocess (POSIX only). Best-effort, not a hard security + boundary -- see execute_code_for_verification()'s docstring.""" + try: + import resource + cpu_s = EXECUTE_CODE_TIMEOUT_S + 2 + resource.setrlimit(resource.RLIMIT_CPU, (cpu_s, cpu_s)) + resource.setrlimit(resource.RLIMIT_AS, (512 * 1024 * 1024, 512 * 1024 * 1024)) + resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) + resource.setrlimit(resource.RLIMIT_NOFILE, (64, 64)) + except Exception: # noqa: BLE001 -- best-effort only, never block verification on this + pass + + +def execute_code_for_verification(code: str, language: str = "python") -> dict: + """v10: harness-internal verification step, called directly by the VERIFY phase below -- NEVER exposed + in TOOL_REGISTRY/TOOL_SCHEMAS, so the model cannot invoke or game it itself; it only ever sees the + resulting pass/fail + stdout/stderr fed back during ERROR_RECOVERY. Best-effort sandbox: this + environment has no spare containerization to isolate a subprocess further, so protection is a short + wall-clock timeout plus POSIX resource limits (CPU time, address space, no core dumps, a reduced file + descriptor count) on an already-unprivileged container user -- not a hard security boundary. Acceptable + because the code executed here is the pipeline's own generated content on trusted infrastructure, not + arbitrary third-party input. Python only for v1; dispatched on `language` so a second language later is + additive, not a rewrite.""" + if language != "python": + return {"passed": False, "stdout": "", + "stderr": f"language '{language}' is not supported for execution-based verification yet " + "(python only in this version)", "exit_code": None, "timed_out": False} + tmp_dir = tempfile.mkdtemp(prefix="gijl_verify_") + try: + script_path = os.path.join(tmp_dir, "candidate.py") + with open(script_path, "w", encoding="utf-8") as f: + f.write(code) + try: + proc = subprocess.run([sys.executable, "-I", "candidate.py"], cwd=tmp_dir, capture_output=True, + text=True, timeout=EXECUTE_CODE_TIMEOUT_S, preexec_fn=_limit_resources) + return {"passed": proc.returncode == 0, "stdout": _trunc(proc.stdout, EXECUTE_CODE_MAX_OUTPUT_CHARS), + "stderr": _trunc(proc.stderr, EXECUTE_CODE_MAX_OUTPUT_CHARS), + "exit_code": proc.returncode, "timed_out": False} + except subprocess.TimeoutExpired as e: + return {"passed": False, "stdout": _trunc(e.stdout or "", EXECUTE_CODE_MAX_OUTPUT_CHARS), + "stderr": _trunc((e.stderr or "") + f"\n[killed: exceeded {EXECUTE_CODE_TIMEOUT_S}s timeout]", + EXECUTE_CODE_MAX_OUTPUT_CHARS), + "exit_code": None, "timed_out": True} + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + +def _grounding_check(proposal: str, research_steps: list) -> dict: + """Deterministic (non-LLM) grounding check for a non-code proposal: does its text actually draw on + content gathered during ACT, rather than being asserted from nothing? A cheap bigram-overlap heuristic + reusing the same _shingles/_norm_text primitives the near-duplicate index already uses -- the harness + decides pass/fail here deterministically, from real recorded tool results, not a vague LLM self-rating.""" + if not research_steps: + return {"passed": False, "reason": "no tool calls were made to ground this against"} + gathered = " ".join(str(s.get("result", "")) for s in research_steps) + prop_sh, gathered_sh = _shingles(proposal), _shingles(gathered) + if not prop_sh: + return {"passed": False, "reason": "proposal too short to check"} + if not gathered_sh: + return {"passed": False, "reason": "no usable text in the gathered tool results"} + overlap = len(prop_sh & gathered_sh) / len(prop_sh) + passed = overlap >= 0.06 + reason = "" if passed else f"proposal shares too little content with the gathered tool results (overlap={overlap:.3f})" + return {"passed": passed, "overlap_ratio": round(overlap, 3), "reason": reason} + + +def _code_syntax_ok(text: str): + """Cheap, real, non-LLM pre-check (DeepMind's synthetic-data finding: verify with code execution where + you can). A syntax check only -- used as a fast path so an unparseable block never reaches the heavier + sandbox in execute_code_for_verification(). Returns None when there is no Python block to check.""" + blocks = _PY_CODE_BLOCK_RE.findall(text or "") + if not blocks: + return None + import ast as _ast + return all(_safe_ast_parse(_ast, b) for b in blocks) + + +def _safe_ast_parse(ast_mod, code: str) -> bool: + try: + ast_mod.parse(code) + return True + except SyntaxError: + return False + except Exception: # noqa: BLE001 -- a parser quirk must never fail the whole record + return True + + +def _agent_task_tool_schemas(base_schemas: list) -> list: + # select_scenario has no meaning once a scenario has already been chosen for this cycle. v10: no + # finish_task_answer tool anymore -- ACT only gathers information; PROPOSE is a separate, dedicated + # phase where the model writes its answer, so there is nothing to "finish" with a tool call here. + return [s for s in base_schemas if s["function"]["name"] != "select_scenario"] + + +THINK_SYSTEM_TEMPLATE = """Reasoning strength: {reasoning}. -AGENT_TASK_SYSTEM_TEMPLATE = """Reasoning strength: {reasoning}. +A real person sent you a task (background material follows, private context for you only -- never quote it +or refer to it in your reply). Before doing anything else: -You are a careful, thinking assistant helping a real person with a real task. You have research tools (listed -below); use them whenever a claim needs checking or a fact needs looking up, rather than guessing from memory. -Use only what the task actually needs -- not every tool, not for its own sake. +1. Restate their question or request in your own words as a short, self-contained paragraph wrapped in + ... tags -- a reader with no other context must understand it fully from this alone (never + mention the material, a scenario, an article or a URL). +2. In a few sentences, say what you will need to check or look up to answer well. +3. On the LAST line, write exactly "NEEDS_TOOLS: yes" if answering this well requires checking something you + cannot be fully certain of from memory alone (a fact, a current detail, specific API/library behavior, a + piece of code that should be run) -- true for most real tasks -- or "NEEDS_TOOLS: no" only for a task that + is genuinely pure opinion/creative/stylistic with nothing to check. -TOOLS (call them when you need them; native format below): +{language_rule} +""" + +ACT_SYSTEM_TEMPLATE = """Reasoning strength: {reasoning}. + +You are investigating this task before answering it: +{task_prompt} +Your own plan: {think} + +Use the research tools below whenever a claim needs checking or a fact needs looking up -- never answer a +checkable question from memory alone. Use only what the task actually needs, not every tool for its own sake. + +TOOLS (native format below): {tools_description} HOW TO CALL A TOOL (several invokes may share one block): @@ -3242,33 +3518,157 @@ HOW TO CALL A TOOL (several invokes may share one block): -BUDGET: at most {max_calls} tool calls total for this task. When you have enough to answer well (often well -before the limit), stop investigating and call finish_task_answer. +BUDGET: at most {max_calls} tool calls total. When you have enough to answer well (often well before the +limit), reply with NO tool call and a one-line note that you are ready to answer. + +{language_rule} +""" + +PROPOSE_SYSTEM_TEMPLATE = """Reasoning strength: {reasoning}. + +Write your complete, final answer to this task now: +{task_prompt} +Your own plan: {think} -FIRST, before any tool call, restate the person's question in your own words as a short, self-contained -paragraph wrapped in ... tags -- a reader with no other context must understand it fully from this -alone (never mention the material, the scenario, a URL, or a source). THEN investigate as needed. When ready, -call finish_task_answer with your complete, accurate, well-structured final answer -- say so plainly if -something could not be verified, instead of guessing. If the task has a risky edge, handle it the way a careful -expert would (a brief warning and a safe default), not a lecture and not silent compliance. +{research_note} + +If your answer includes code, put it in a single fenced ```python code block -- it WILL be executed for real +to verify it before anything else happens, so make it runnable as-is (no placeholders, no "...", include any +needed imports), not just a sketch. If it makes factual/technical claims, make sure they are grounded in what +you found above, not guessed. Do not pad the answer with meta-commentary about your process -- just the +answer a real person would receive. {language_rule} """ +RECOVERY_SEARCH_SYSTEM_TEMPLATE = """Reasoning strength: {reasoning}. -def _agent_task_tool_schemas(base_schemas: list) -> list: - # finish_task_answer replaces select_scenario as the terminal action for this loop -- the two must never be - # offered together (select_scenario has no meaning once a scenario has already been chosen for this cycle). - return [s for s in base_schemas if s["function"]["name"] != "select_scenario"] + [AGENT_TASK_FINISH_SCHEMA] +Your previous answer to this task did not verify: +{task_prompt} + +What went wrong: +{error_summary} + +Use the research tools below to search for this exact error, or an already-solved instance of this kind of +problem, in real external sources. Do not guess a fix from memory -- find real evidence first. + +TOOLS (native format below): +{tools_description} + +HOW TO CALL A TOOL: + + +{example_query} + + +BUDGET: at most {max_calls} searches. When you have enough to propose a fix, reply with NO tool call. -def build_agent_task_prompt(plan: CyclePlan, base_schemas: list) -> str: +{language_rule} +""" + +RECOVERY_FIX_TEMPLATE = """Based on what you just found, write ONE consolidated fix for this task that does +not conflict with your original approach -- reconcile the candidate fixes you found into a single coherent +answer rather than patching in every suggestion. Before the fix itself, write one line starting "RATIONALE:" +explaining briefly why this is the right fix given what you found. Then give the complete, corrected final +answer (same format rules as before: a single runnable ```python block if there is code, no placeholders). + +{task_prompt} + +{language_rule} +""" + +REVIEW_SYSTEM_TEMPLATE = """Reasoning strength: {reasoning}. + +This is your final check before this answer is delivered: +{task_prompt} + +{proposal} + + +Does this fully answer the task? Anything left ungrounded or any loose end? {verification_note} + +Write a short review note, then on the LAST line write exactly "REVISION: none" if the answer above should be +delivered as-is, or "REVISION: " if it needs one small fix -- this is your only +chance to revise, so only do it if something is genuinely wrong or missing. + +{language_rule} +""" + + +def _run_verification(client, proposal: str, research_steps: list, task_prompt: str, think: str) -> dict: + """Harness-selected, not model-selected: picks the verification method from what `proposal` actually + contains, runs it for real, and returns {"method", "passed", "evidence"}. This is what makes + "verify everything you produce" generalize across arbitrary task kinds instead of one fixed routine.""" + code_blocks = _PY_CODE_BLOCK_RE.findall(proposal) + if code_blocks: + if _code_syntax_ok(proposal) is False: + return {"method": "code_execution", "passed": False, + "evidence": {"stderr": "SyntaxError: the proposed code does not even parse", "stdout": "", + "exit_code": None, "timed_out": False}} + exec_result = execute_code_for_verification(code_blocks[0].strip()) + return {"method": "code_execution", "passed": exec_result["passed"], + "evidence": {k: exec_result[k] for k in ("stdout", "stderr", "exit_code", "timed_out")}} + if research_steps: + g = _grounding_check(proposal, research_steps) + return {"method": "grounding", "passed": g["passed"], "evidence": g} + # self_consistency: the one branch that still asks the model, deliberately -- logged as a real verify + # step either way, so "always verify what you produce" is visible in every trajectory, per the + # requirement that this habit must not be conditional on the task happening to involve code or search. + sc_messages = [{"role": "system", "content": "Reasoning strength: low.\n\nYou wrote an answer to a task. " + "Re-read it against your own plan and reply with exactly one word, PASS or FAIL, then a " + "one-line reason."}, + {"role": "user", "content": f"{task_prompt}\n{think}\n" + f"{proposal}"}] + sc_msg = client.chat(sc_messages, max_new_tokens=120, temperature=0.2) + sc_text = _strip_atem_noise(sc_msg.get("content", "")) + sc_passed = bool(re.search(r"\bPASS\b", sc_text, re.IGNORECASE)) and not re.search(r"\bFAIL\b", sc_text, re.IGNORECASE) + return {"method": "self_consistency", "passed": sc_passed, "evidence": {"reply": _trunc(sc_text, 300)}} + + +def _run_error_recovery(client, plan: CyclePlan, act_schemas: list, available: set, task_prompt: str, + error_summary: str, reasoning: str, lang_rule: str) -> dict: + """ERROR_RECOVERY: real tool calls to search external sources for this exact error / an already-solved + instance of the problem, then one call that reconciles whatever was found into a single consolidated + fix. Bounded by AGENT_TASK_MAX_RECOVERY_SEARCHES. Always returns a dict (never raises) so a flaky + recovery call degrades to "no fix found" rather than losing the whole attempt.""" tool, query = _example_call(plan) - schemas = _agent_task_tool_schemas(base_schemas) - return AGENT_TASK_SYSTEM_TEMPLATE.format( - tools_description=_render_tool_descriptions(schemas), example_tool=tool, example_query=query, - max_calls=AGENT_TASK_MAX_TOOL_CALLS, reasoning=REASONING_SCOUT, - language_rule=_LANG_RULE.format(language=plan.language)) + recovery_messages = [{"role": "system", "content": RECOVERY_SEARCH_SYSTEM_TEMPLATE.format( + reasoning=reasoning, task_prompt=task_prompt, error_summary=error_summary, + tools_description=_render_tool_descriptions(act_schemas), example_tool=tool, example_query=query, + max_calls=AGENT_TASK_MAX_RECOVERY_SEARCHES, language_rule=lang_rule)}, + {"role": "user", "content": "Search now."}] + external_search_steps, used = [], 0 + while used < AGENT_TASK_MAX_RECOVERY_SEARCHES: + rmsg = client.chat(recovery_messages, max_new_tokens=TOOL_TURN_MAX_NEW_TOKENS, tools=act_schemas) + rcalls = _extract_tool_calls(rmsg.get("content", ""))[:MAX_CALLS_PER_TURN] + rclean = _strip_atem_noise(rmsg.get("content", "")) + if not rcalls: + break + recovery_messages.append({"role": "assistant", "content": rclean}) + for call in rcalls: + if used >= AGENT_TASK_MAX_RECOVERY_SEARCHES: + break + used += 1 + name, args = call["name"], call.get("args") or {} + result = (call_tool(name, args) if name in available and name in TOOL_REGISTRY + else {"error": f"tool '{name}' is unavailable"}) + print(f"[agent_task cycle {plan.cycle}] RECOVERY {name}(" + f"{_trunc(json.dumps(args, ensure_ascii=False), 100)}) -> {_result_brief(result)}", flush=True) + text = _render_tool_result(result) + recovery_messages.append({"role": "tool", "name": name, "content": text}) + external_search_steps.append({"tool": name, + "arguments_json": json.dumps(args, ensure_ascii=False, default=str), + "result": _trunc(text, TRACE_RESULT_CHARS)}) + fix_messages = recovery_messages + [{"role": "user", "content": RECOVERY_FIX_TEMPLATE.format( + task_prompt=task_prompt, language_rule=lang_rule)}] + fmsg = client.chat(fix_messages, max_new_tokens=PROPOSE_MAX_NEW_TOKENS) + fix_text = _clean_field(_strip_atem_noise(fmsg.get("content", ""))) + rationale_m = _RATIONALE_RE.search(fix_text) + return {"error_summary": error_summary, "external_search_steps": external_search_steps, + "candidate_fixes": [s["result"][:300] for s in external_search_steps], + "chosen_fix_rationale": _trunc(rationale_m.group(1), 300) if rationale_m else "", + "fix_text": _RATIONALE_RE.sub("", fix_text).strip() or fix_text} def build_agent_task_kickoff(scenario: dict) -> str: @@ -3280,80 +3680,69 @@ def build_agent_task_kickoff(scenario: dict) -> str: "tags, then help them.") -def _code_syntax_ok(text: str): - """Cheap, real, non-LLM verification signal (DeepMind's synthetic-data finding: verify with code execution - where you can, rather than trusting an LLM judge for everything). A syntax check only -- never executes - anything -- so it is safe to run unconditionally. Returns None when there is no Python block to check.""" - blocks = _PY_CODE_BLOCK_RE.findall(text or "") - if not blocks: - return None - import ast as _ast - return all(_safe_ast_parse(_ast, b) for b in blocks) - - -def _safe_ast_parse(ast_mod, code: str) -> bool: - try: - ast_mod.parse(code) - return True - except SyntaxError: - return False - except Exception: # noqa: BLE001 -- a parser quirk must never fail the whole record - return True - - def run_agent_task(client, plan: CyclePlan, scenario: dict) -> dict: - """Stage 2 for loops with uses_tools=True: a bounded ReAct loop over the SAME tool registry as scouting, but - solving a genuine (simulated) user task instead of looking for raw material. Returns the same {"ok", "records"} - shape as every other builder (generate_records()'s other callers, verify_generated(), write_records() all stay - unchanged).""" + """v10: THINK -> ACT -> PROPOSE -> VERIFY -> (ERROR_RECOVERY -> PROPOSE -> VERIFY)* -> REVIEW -> OUTPUT, + each phase driven by the harness rather than one free-running system prompt. Returns the same + {"ok", "records"} shape as every other builder (generate_records()'s other callers, verify_generated(), + write_records() all stay unchanged).""" language = plan.language + reasoning = REASONING_SCOUT + lang_rule = _LANG_RULE.format(language=language) base_schemas = available_tool_schemas() - schemas = _agent_task_tool_schemas(base_schemas) - available = {s["function"]["name"] for s in schemas} - messages = [{"role": "system", "content": build_agent_task_prompt(plan, base_schemas)}, - {"role": "user", "content": build_agent_task_kickoff(scenario)}] - trace = TraceRecorder(plan, "", sorted(available), data_type="agent_trajectory", role="agent_task") - used = nudges = 0 - task_prompt, final_answer, confidence, outcome = "", "", "medium", "no_final_answer" + act_schemas = _agent_task_tool_schemas(base_schemas) + available = {s["function"]["name"] for s in act_schemas} + + # --- THINK --------------------------------------------------------------------------------------------- + think_messages = [{"role": "system", "content": THINK_SYSTEM_TEMPLATE.format(reasoning=reasoning, language_rule=lang_rule)}, + {"role": "user", "content": build_agent_task_kickoff(scenario)}] + msg = client.chat(think_messages, max_new_tokens=THINK_MAX_NEW_TOKENS) + if msg.get("error") == "context_too_long": + return _fail("context_too_long") + think_raw = _strip_atem_noise(msg.get("content", "")) + m = _TASK_TAG_RE.search(think_raw) + task_prompt = _clean_field(m.group(1)) if m else "" + needs_tools_m = _NEEDS_TOOLS_RE.search(think_raw) + needs_tools = True if not needs_tools_m else needs_tools_m.group(1).lower() == "yes" + think = _clean_field(_NEEDS_TOOLS_RE.sub(" ", _TASK_TAG_RE.sub(" ", think_raw)))[:1500] + if not task_prompt: + task_prompt = _clean_field(scenario.get("title", "")) or "(the agent never restated the task -- see think)" + + # --- ACT ------------------------------------------------------------------------------------------------- + research_steps, used, nudges = [], 0, 0 + tool, query = _example_call(plan) + act_messages = [{"role": "system", "content": ACT_SYSTEM_TEMPLATE.format( + reasoning=reasoning, task_prompt=task_prompt, think=think or "(no plan given)", + tools_description=_render_tool_descriptions(act_schemas), example_tool=tool, example_query=query, + max_calls=AGENT_TASK_MAX_TOOL_CALLS, language_rule=lang_rule)}, + {"role": "user", "content": "Begin. Call a tool if you need to, or say you are ready to answer."}] + act_outcome = None while used < AGENT_TASK_MAX_TOOL_CALLS: - msg = client.chat(messages, max_new_tokens=TOOL_TURN_MAX_NEW_TOKENS, tools=schemas) + msg = client.chat(act_messages, max_new_tokens=TOOL_TURN_MAX_NEW_TOKENS, tools=act_schemas) if msg.get("error") == "context_too_long": - outcome = "context_too_long" + act_outcome = "context_too_long" break if (msg.get("error") or msg.get("timed_out")) and not (msg.get("content") or "").strip(): - outcome = "backend_error" + act_outcome = "backend_error" break raw = msg.get("content", "") calls = _extract_tool_calls(raw)[:MAX_CALLS_PER_TURN] clean = _strip_reasoning_leak(_strip_atem_noise(raw)) - if not task_prompt: - m = _TASK_TAG_RE.search(raw) - if m: - task_prompt = _clean_field(m.group(1)) if not calls: + if research_steps or not needs_tools: + break # a natural stop: it already researched something, or said it would not need to if nudges < AGENT_TASK_MAX_NUDGES: nudges += 1 - messages += [{"role": "assistant", "content": clean or "(empty reply)"}, - {"role": "user", "content": "Call finish_task_answer with your final answer, or call " - "one of the research tools if you still need to check something."}] + act_messages += [{"role": "assistant", "content": clean or "(empty reply)"}, + {"role": "user", "content": "You said this task needs checking -- call a " + "research tool now, or explain briefly why none actually applies."}] continue - outcome = "no_tool_calls" - break - messages.append({"role": "assistant", "content": clean}) - step, done = [], False + break # out of nudges; ACT ends with whatever it has (handled by the needs_tools check below) + act_messages.append({"role": "assistant", "content": clean}) for call in calls: if used >= AGENT_TASK_MAX_TOOL_CALLS: break used += 1 name, args = call["name"], call.get("args") or {} - if name == "finish_task_answer": - final_answer = _clean_field(args.get("final_answer", "")) - confidence = str(args.get("confidence", "medium")).strip().lower() - confidence = confidence if confidence in ("low", "medium", "high") else "medium" - outcome = "success" if final_answer else "empty_final_answer" - done = True - step.append((name, {"confidence": confidence}, "(final answer recorded)", "model")) - break if name not in available or name not in TOOL_REGISTRY: result = {"error": f"tool '{name}' is unavailable"} else: @@ -3362,34 +3751,95 @@ def run_agent_task(client, plan: CyclePlan, scenario: dict) -> dict: if isinstance(result, dict) and "error" not in result: result = {**result, "calls_left": left} if left <= BUDGET_WARN_AT: - result["next_step"] = f"only {left} tool call(s) left: wrap up and call finish_task_answer" - print(f"[agent_task cycle {plan.cycle}] {name}({_trunc(json.dumps(args, ensure_ascii=False), 100)}) " + result["next_step"] = f"only {left} tool call(s) left: wrap up soon" + print(f"[agent_task cycle {plan.cycle}] ACT {name}({_trunc(json.dumps(args, ensure_ascii=False), 100)}) " f"-> {_result_brief(result)}", flush=True) text = _render_tool_result(result) - messages.append({"role": "tool", "name": name, "content": text}) - step.append((name, args, text, "model")) - if step: - trace.add_step(clean, step) - if done: + act_messages.append({"role": "tool", "name": name, "content": text}) + research_steps.append({"tool": name, "arguments_json": json.dumps(args, ensure_ascii=False, default=str), + "result": _trunc(text, TRACE_RESULT_CHARS)}) + if act_outcome in ("context_too_long", "backend_error"): + return _fail(f"agent_task ACT phase failed: {act_outcome}") + if needs_tools and not research_steps: + return _fail("agent_task: THINK said this task needed checking but ACT made no real tool calls " + "(never rely on memory alone)") + + # --- PROPOSE / VERIFY / ERROR_RECOVERY ---------------------------------------------------------------- + research_note = (("Here is what you found:\n" + "\n".join( + f"- {s['tool']}({s['arguments_json']}) -> {s['result'][:400]}" for s in research_steps)) + if research_steps else "You judged that nothing needed checking for this task.") + propose_messages = [{"role": "system", "content": PROPOSE_SYSTEM_TEMPLATE.format( + reasoning=reasoning, task_prompt=task_prompt, think=think or "(no plan given)", + research_note=research_note, language_rule=lang_rule)}, + {"role": "user", "content": "Write your final answer now."}] + msg = client.chat(propose_messages, max_new_tokens=PROPOSE_MAX_NEW_TOKENS) + if msg.get("error") == "context_too_long": + return _fail("context_too_long") + proposal = _clean_field(_strip_atem_noise(msg.get("content", ""))) + if not proposal: + return _fail("agent_task: PROPOSE produced no content") + + attempts, verify_passed, attempt_n = [], False, 1 + while True: + verification = _run_verification(client, proposal, research_steps, task_prompt, think) + attempt_rec = {"attempt": attempt_n, "proposal": proposal, "verification": verification} + if verification["passed"] or attempt_n >= AGENT_TASK_MAX_VERIFY_ATTEMPTS: + attempts.append(attempt_rec) + verify_passed = verification["passed"] break - if not task_prompt: - task_prompt = _clean_field(scenario.get("title", "")) or "(the agent never restated the task -- see steps)" - rec = trace.build(outcome, []) - rec["task_prompt"], rec["final_answer"], rec["confidence"] = task_prompt, final_answer, confidence - rec["source_url"], rec["source_kind"] = scenario["url"], scenario["kind"] - rec["source_license"] = _license_hint(scenario["url"], scenario["kind"]) - rec["verified"], rec["verifier_score"] = "unverified", 0 - code_ok = _code_syntax_ok(final_answer) - if code_ok is not None: - rec["code_syntax_ok"] = code_ok # cheap, real, non-LLM signal -- see _code_syntax_ok - if outcome != "success" or not final_answer: - return _fail(f"agent_task did not finish cleanly: {outcome}") - limits = {"task_prompt": (15, 2000), "final_answer": (30, 8000)} - err = _validate_fields({"task_prompt": task_prompt, "final_answer": final_answer}, language, limits) + error_summary = (verification["evidence"].get("stderr") or verification["evidence"].get("reason") + or verification["evidence"].get("reply") or "verification failed")[:600] + recovery = _run_error_recovery(client, plan, act_schemas, available, task_prompt, error_summary, + reasoning, lang_rule) + attempt_rec["error_recovery"] = {k: v for k, v in recovery.items() if k != "fix_text"} + attempts.append(attempt_rec) + proposal = recovery["fix_text"] or proposal # keep the old proposal if the fix call produced nothing + attempt_n += 1 + + # --- REVIEW ---------------------------------------------------------------------------------------------- + verification_note = ("Every verification attempt passed." if verify_passed else + f"Verification never fully passed after {len(attempts)} attempt(s) -- " + f"last issue: {attempts[-1]['verification']['evidence']}") + review_messages = [{"role": "system", "content": REVIEW_SYSTEM_TEMPLATE.format( + reasoning=reasoning, task_prompt=task_prompt, proposal=proposal, verification_note=verification_note, + language_rule=lang_rule)}, {"role": "user", "content": "Review now."}] + rmsg = client.chat(review_messages, max_new_tokens=REVIEW_MAX_NEW_TOKENS) + review_raw = _strip_atem_noise(rmsg.get("content", "")) + rev_m = _REVISION_RE.search(review_raw) + final_text = proposal + if rev_m: + revision = _clean_field(rev_m.group(1)) + if revision and revision.strip().lower() not in ("none", "none.", "لا شيء"): + final_text = revision + review_note = _clean_field(_REVISION_RE.sub(" ", review_raw))[:800] + + # --- OUTPUT + record assembly ------------------------------------------------------------------------------ + if _PY_CODE_BLOCK_RE.search(final_text): + output_type = "code" + elif research_steps: + output_type = "search_results" + else: + output_type = "text" + limits = {"task_prompt": (15, 2000), "final_text": (30, 8000)} + err = _validate_fields({"task_prompt": task_prompt, "final_text": final_text}, language, limits) if err: return _fail(err) - if _mentions_source(task_prompt) or _mentions_source(final_answer): - return _fail("task_prompt/final_answer refers to the background material instead of being self-contained") + if _mentions_source(task_prompt) or _mentions_source(final_text): + return _fail("task_prompt/output refers to the background material instead of being self-contained") + outcome = "success" if verify_passed else "verify_exhausted" + trainable_for = ["final_answer", "trajectory_discipline"] if verify_passed else ["trajectory_discipline"] + rid = _rid("agent", plan.loop.name, str(plan.cycle), task_prompt[:100], str(time.time())) + rec = {**_envelope("agent_trajectory", plan.loop.name, rid, language, scenario, task_prompt, plan), + "role": "agent_task", "cycle": plan.cycle, "task_prompt": task_prompt, "think": think, + "research_steps": research_steps, "attempts": attempts, "review": review_note, + "output": {"type": output_type, "content": final_text}, + "num_verify_attempts": len(attempts), + "recovered_from_error": any(a.get("error_recovery") for a in attempts), + "outcome": outcome, "trainable": verify_passed, "trainable_for": trainable_for, + "failure_reason": "" if verify_passed else outcome} + # outcome="verify_exhausted" is kept, not discarded -- a trajectory that honestly ran out of verification + # attempts is still a good example of the think/act/verify/error-recovery DISCIPLINE even when the final + # answer itself should not be trained on (trainable_for reflects exactly that split). return {"ok": True, "records": [("agent_trajectory", rec, rec["id"])]} @@ -3409,8 +3859,12 @@ _VERIFY_BOOLS = { "reasoning_qa": ("needs_reasoning", "self_contained", "steps_support_answer", "no_unsupported_claims"), # v6: Constitutional AI's finding (self-critique against explicit NAMED principles beats a vague verdict) # applied to agent behavior instead of only harmlessness -- see the v6 changelog at the top of this file. - "agent_task": ("efficient_steps", "grounded_tool_args", "verified_before_claiming", "calibrated_response", - "self_contained_task"), + # v10: "verified_before_claiming" split into "real_verification_used" (did a genuine, harness-run VERIFY + # attempt happen at all) + "honest_about_limits" (does the output say so when verification never fully + # passed) -- now that verification is a real, recorded, code-enforced phase (see run_agent_task()) rather + # than something the judge had to infer, these can be checked against the actual attempts list. + "agent_task": ("efficient_steps", "grounded_tool_args", "real_verification_used", "calibrated_response", + "self_contained_task", "honest_about_limits"), } _VERIFY_HEAD = ("Reasoning strength: low.\n\nYou are an INDEPENDENT reviewer of ONE training example. You did not write it. " "Judge it like a strict dataset editor: when unsure, fail it.\n\n{untrusted_note}\n\n") @@ -3458,25 +3912,33 @@ Fields: Respond with ONLY a JSON object: {{"needs_reasoning": true, "self_contained": true, "steps_support_answer": true, "no_unsupported_claims": true, "verdict": "pass", "score": 4, "issues": []}} """, - "agent_task": _VERIFY_HEAD + """Task: a tool-using agent trajectory. You see the person's TASK, the agent's TOOL STEPS -(calls + results, in order), and its FINAL ANSWER. Judge the trajectory AS A WHOLE, not just the final text. + "agent_task": _VERIFY_HEAD + """Task: a code-enforced tool-using agent trajectory with explicit +think -> act -> propose -> verify -> (error-recovery) -> review -> output phases, every phase driven by the +harness, not left to the model's own discretion. You see the person's TASK, the agent's THINK plan, its TOOL +STEPS (research calls + results, in order), its VERIFICATION ATTEMPTS (each attempt's method -- code_execution, +grounding, or self_consistency -- and whether it passed, plus any error-recovery search + consolidated fix after +a failed attempt), its REVIEW note, and its final OUTPUT. Judge the trajectory AS A WHOLE, not just the final text. Fields: -- "efficient_steps": true only if the agent did not waste calls (no repeated/near-identical calls, no calls - irrelevant to the task, no "double-checking" past the point of usefulness). +- "efficient_steps": true only if the agent did not waste research calls (no repeated/near-identical calls, no + calls irrelevant to the task, no "double-checking" past the point of usefulness). - "grounded_tool_args": true only if every tool argument was a specific, well-formed query for THIS task (not vague, not copy-pasted from the task verbatim with no refinement). -- "verified_before_claiming": true only if the agent checked a checkable claim with a tool instead of asserting - it from memory, OR there was genuinely nothing in this task that needed checking. +- "real_verification_used": true only if the VERIFICATION ATTEMPTS list shows a genuine attempt whose method + actually fits what was proposed (code got code_execution, a factual/research claim got grounding), OR there + was truly nothing in this task that needed checking (a pure opinion/creative task). - "calibrated_response": true only if the final answer is appropriately careful on any risky edge (a brief warning and a safe default, not silent compliance and not a refusal where none was warranted). - "self_contained_task": true only if the TASK text never refers to "the material", "the scenario", a URL, or a source -- it must read as a real person's question on its own. +- "honest_about_limits": true only if, whenever a verification attempt in VERIFICATION ATTEMPTS never fully + passed, the final OUTPUT says so plainly instead of presenting the unverified content with full confidence. + Automatically true if every attempt passed. - "verdict": "pass" or "fail". "score": integer 1-5. "issues": short strings, empty list if none. Respond with ONLY a JSON object: -{{"efficient_steps": true, "grounded_tool_args": true, "verified_before_claiming": true, "calibrated_response": true, - "self_contained_task": true, "verdict": "pass", "score": 4, "issues": []}} +{{"efficient_steps": true, "grounded_tool_args": true, "real_verification_used": true, "calibrated_response": true, + "self_contained_task": true, "honest_about_limits": true, "verdict": "pass", "score": 4, "issues": []}} """, } @@ -3497,13 +3959,25 @@ def _verify_input(plan: CyclePlan, scenario: dict, rec: dict): m = rec["messages"] return f"\n{m[0]['content']}\n\n\n\n{m[1]['content']}\n\n\n{ctx}", None if loop == "agent_task": - lines = [] - for i, s in enumerate(rec.get("steps", []), 1): - for c, r in zip(s.get("tool_calls", []), s.get("tool_results", [])): - lines.append(f"{i}. {c['name']}({c['arguments_json']}) -> {r['content'][:400]}") + # v10: the new phase-machine shape (think/research_steps/attempts/review/output) replaces the old flat + # steps/final_answer fields -- see the record assembly at the end of run_agent_task(). + lines = [f"{i}. {s.get('tool')}({s.get('arguments_json')}) -> {str(s.get('result'))[:400]}" + for i, s in enumerate(rec.get("research_steps") or [], 1)] steps_text = "\n".join(lines) or "(no tool calls)" - return (f"\n{rec['task_prompt']}\n\n\n\n{steps_text}\n\n\n" - f"\n{rec['final_answer']}\n"), None + att_lines = [] + for a in rec.get("attempts") or []: + v = a.get("verification") or {} + att_lines.append(f"attempt {a.get('attempt')}: method={v.get('method')} passed={v.get('passed')}") + er = a.get("error_recovery") + if er: + att_lines.append(f" error_recovery: {str(er.get('error_summary', ''))[:200]} | " + f"chosen_fix: {str(er.get('chosen_fix_rationale', ''))[:200]}") + attempts_text = "\n".join(att_lines) or "(no verification attempts)" + output_text = (rec.get("output") or {}).get("content", "") + return (f"\n{rec.get('task_prompt', '')}\n\n\n\n{rec.get('think', '')}\n\n\n" + f"\n{steps_text}\n\n\n\n{attempts_text}\n" + f"\n\n\n{rec.get('review', '')}\n\n\n" + f"\n{output_text}\n"), None steps = "\n".join(f"{i}. {t}" for i, t in enumerate(rec["reasoning_steps"], 1)) return (f"\n{rec['question']}\n\n\n\n{steps}\n\n\n" f"\n{rec['answer']}\n\n\n\n{(scenario.get('body') or '')[:2000]}\n"), None @@ -3594,12 +4068,67 @@ def _atomic_write_json(path: str, obj): os.replace(tmp, path) # a crash mid-write can no longer corrupt seen.json / status.json +# v10 anonymization: `generated_by` (exact model id) and `collector_tag` (which worker) used to be on every +# record and were how the README/manifest attributed a row to a teacher model. The user asked that nothing +# public (records or the README card) name which model produced a row -- MODEL_TAG keeps being used for +# purely internal/local things (STATE_DIR, the shard filename derivation, the in-memory dedup index) since +# that use was explicitly exempted ("يستثنى cache"); it is simply never written into a record or the README. +# +# v10 schema-consistency fix for the Hub dataset viewer: the old per-data_type/per-split files already kept +# each file's rows homogeneous by construction; now that every data_type shares ONE physical file +# (SHARD_PATH locally, MERGED_PATH/MERGED_REPO_PATH once synced), a row from one data_type must not have a +# different column set than a row from another, or the Hub's schema inference can fail outright (see the +# historical comment this replaced, previously on _backfill: "Mixed key sets inside one JSONL break the +# Hub's dataset viewer / `datasets` loader"). ALL_RECORD_FIELDS is the union of every field any builder +# below ever sets; _normalize_envelope_fields() fills whatever a given row's builder did not set with None, +# so every row -- of every data_type -- ends up with identical top-level columns. +_SHARED_ENVELOPE_FIELDS = ("id", "schema_version", "data_type", "loop", "language", "split", + "content_hash", "source_kind", "source_url", "generated_at", "run_id", "focus", + "source_license", "source_chars", "verified", "verifier_score", "trainable_for") +_TYPE_SPECIFIC_FIELDS = ( + "category", "subcategory_note", "risk_severity", "response_pattern", "prompt", "chosen", "rejected", + "judgment_rationale", "chosen_chars", "rejected_chars", + "parent_id", "origin", "domain", "difficulty", "messages", + "question", "reasoning_steps", "answer", "confidence", + "role", "cycle", "goal", "tools_offered", "steps", "num_tool_calls", "outcome", + "records_written", "origin_mix", "elapsed_s", "trainable", "failure_reason", + "n_error_results", "reasoning_spill", + "task_prompt", "think", "research_steps", "attempts", "review", "output", + "num_verify_attempts", "recovered_from_error", + "title", "chars", "body_sha256", "status", "error", "body_excerpt", + # kimi-only data_type ("cross_review", see kimi_extra() in build_workers.py) -- included here, not just + # in that worker's own file, so every OTHER worker's rows also normalize these to None instead of the + # merged file ending up with a data_type-dependent column set again (see the note above this tuple). + # No "reviewed_tag" -- reviewed_id alone (a sha256 hash, globally unique) identifies the reviewed row + # without storing which worker wrote it, per this project's anonymization requirement. + "reviewed_id", "reviewed_type", "reviewed_verified", "verdict", "score", "issues") +ALL_RECORD_FIELDS = _SHARED_ENVELOPE_FIELDS + _TYPE_SPECIFIC_FIELDS +_ALL_RECORD_FIELDS_SET = set(ALL_RECORD_FIELDS) + + +def _normalize_envelope_fields(rec: dict) -> dict: + # v10: STRICT normalization, not just filling gaps -- a stray key from a malformed/legacy-shaped row + # (e.g. an old "final_answer"/"steps"-only agent_trajectory row that predates this envelope) must not + # survive into the shard, or it reintroduces exactly the per-row column drift this envelope exists to + # prevent (see the note above ALL_RECORD_FIELDS). Any key outside ALL_RECORD_FIELDS is dropped; every key + # inside it is guaranteed present, defaulting to None. + for k in [k for k in rec if k not in _ALL_RECORD_FIELDS_SET]: + del rec[k] + for k in ALL_RECORD_FIELDS: + rec.setdefault(k, None) + return rec + + def append_record(data_type: str, record: dict, split_id: str = None): - folder = type_dir(data_type) - os.makedirs(folder, exist_ok=True) - path = os.path.join(folder, f"{split_for_record(record, split_id)}.jsonl") + """v10: every data_type now writes into the SAME single shard file (SHARD_PATH) -- `split` is a row + column, not a folder/filename choice. Still one writer per physical file (this worker only ever + appends to its own shard), so the existing flock-based atomic append stays exactly as race-free as + the old per-type-folder layout.""" + record["split"] = split_for_record(record, split_id) + _normalize_envelope_fields(record) + os.makedirs(SHARD_DIR, exist_ok=True) line = json.dumps(record, ensure_ascii=False) + "\n" - with open(path, "a", encoding="utf-8") as f: + with open(SHARD_PATH, "a", encoding="utf-8") as f: if fcntl: fcntl.flock(f.fileno(), fcntl.LOCK_EX) try: @@ -3653,12 +4182,14 @@ class DedupIndex: self.sims[dt].append(sim) def load(self): - for dt in DATA_TYPES: - for split in SPLITS: - for rec in _read_jsonl(os.path.join(type_dir(dt), f"{split}.jsonl")): - self._index(dt, rec) - if dt in _NEAR_DUP_TYPES and rec.get("source_url"): - SOURCE_SPLITS.setdefault(_norm_source_key(rec["source_url"]), split) + # v10: one physical file for this worker instead of DATA_TYPES x SPLITS nested folders. + for rec in _load_shard_records(): + dt, split = rec.get("data_type"), rec.get("split") + if dt not in DATA_TYPES: + continue + self._index(dt, rec) + if dt in _NEAR_DUP_TYPES and rec.get("source_url") and split in SPLITS: + SOURCE_SPLITS.setdefault(_norm_source_key(rec["source_url"]), split) def dup_reason(self, dt, rec) -> str: if _dedup_key(dt, rec) in self.keys[dt]: @@ -3714,15 +4245,24 @@ def load_status() -> dict: return {} -def _count_lines(path) -> int: - if not os.path.isfile(path): - return 0 - with open(path, encoding="utf-8") as f: - return sum(1 for _ in f) +def _load_shard_records(path: str = None) -> list: + """v10: read the one physical worker shard (or, during a sync merge, an already-concatenated + file) once, as plain dicts. Replaces the old DATA_TYPES x SPLITS nested-file iteration every + counting/reporting function used to do.""" + return list(_read_jsonl(path or SHARD_PATH)) + + +def _group_by_type_split(records: list) -> dict: + counts = {dt: {s: 0 for s in SPLITS} for dt in DATA_TYPES} + for rec in records: + dt, sp = rec.get("data_type"), rec.get("split") + if dt in counts and sp in SPLITS: + counts[dt][sp] += 1 + return counts def write_status(cycle: int, state: RunState): - type_counts = {dt: {s: _count_lines(os.path.join(type_dir(dt), f"{s}.jsonl")) for s in SPLITS} for dt in DATA_TYPES} + type_counts = _group_by_type_split(_load_shard_records()) _atomic_write_json(STATUS_PATH, { "cycle": cycle, "split_counts": type_counts["preference_pair"], # legacy shape (DPO only) "type_counts": type_counts, "stats": dict(state.stats), "updated_at": _now()}) @@ -3747,7 +4287,7 @@ def write_records(state: RunState, items: list): state.dedup.add(dt, rec) written[dt] += 1 if dt != "source_index": - print(f"[write] {dt} id={rec['id']} -> {DATA_TYPES[dt]['folder']}/{split_for_record(rec, split_id)}.jsonl " + print(f"[write] {dt} id={rec['id']} split={rec.get('split')} -> {os.path.basename(SHARD_PATH)} " f"(total {len(state.dedup.keys[dt])})", flush=True) return written, None @@ -3756,26 +4296,33 @@ _TALLY_FIELDS = ("language", "category", "domain", "risk_severity", "response_pa "origin", "difficulty", "confidence", "outcome", "status", "verified", "focus") -def build_manifest() -> dict: - manifest = {"generated_at": _now(), "schema_version": SCHEMA_VERSION, "model": MODEL_ID, +def build_manifest(records: list = None) -> dict: + """v10: anonymized (no `model`/MODEL_ID field -- see the v10 storage/anonymization note at + append_record()) and built from an in-memory record list -- the worker's own shard for a local/ + offline refresh, or the full cross-worker merge when called from sync_to_hf().""" + records = _load_shard_records() if records is None else records + manifest = {"generated_at": _now(), "schema_version": SCHEMA_VERSION, "repo": TARGET_DATASET_REPO, "data_types": {}} + by_type = {dt: [] for dt in DATA_TYPES} + for rec in records: + if rec.get("data_type") in by_type: + by_type[rec["data_type"]].append(rec) for dt, meta in DATA_TYPES.items(): - info = {"folder": meta["folder"], "title": meta["title"], "description": meta["description"], - "splits": {}, "total": 0} + rows = by_type[dt] + info = {"title": meta["title"], "description": meta["description"], + "splits": {s: 0 for s in SPLITS}, "total": 0} tallies = {f: Counter() for f in _TALLY_FIELDS} - for split in SPLITS: - n = 0 - for rec in _read_jsonl(os.path.join(type_dir(dt), f"{split}.jsonl")): - n += 1 - for f in _TALLY_FIELDS: - v = rec.get(f) - if isinstance(v, str) and v: - tallies[f][v] += 1 - info["splits"][split] = n - info["total"] += n + for rec in rows: + if rec.get("split") in SPLITS: + info["splits"][rec["split"]] += 1 + info["total"] += 1 + for f in _TALLY_FIELDS: + v = rec.get(f) + if isinstance(v, str) and v: + tallies[f][v] += 1 info["tallies"] = {f: dict(c.most_common()) for f, c in tallies.items() if c} manifest["data_types"][dt] = info - manifest["audit"] = build_audit() + manifest["audit"] = build_audit(records) _atomic_write_json(MANIFEST_PATH, manifest) return manifest @@ -3795,83 +4342,90 @@ def _count_error_results(steps) -> int: if str(r.get("content", "")).lstrip().startswith('{"error"')) -def _backfill(dt: str, rec: dict) -> bool: - """Add the v5 columns to an older row. Mixed key sets inside one JSONL break the Hub's dataset viewer / - `datasets` loader, so every row of a config must carry the same columns.""" +def _backfill(rec: dict) -> bool: + """v10: normalizes a row written under an older schema -- strips the removed generated_by/ + collector_tag identifying fields (see the anonymization note at append_record()) and fills in + any column the superset envelope now expects (_normalize_envelope_fields) so it merges safely + with current rows in the one shared file. Mixed key sets across rows of the same file is exactly + what breaks the Hub's dataset viewer / `datasets` loader -- this is the guard against that.""" before = json.dumps(rec, sort_keys=True, ensure_ascii=False) - url = rec.get("source_url", "") or "" + rec.pop("generated_by", None) + rec.pop("collector_tag", None) + url = rec.get("source_url") or "" rec.setdefault("run_id", "legacy") rec.setdefault("source_license", _license_hint(url, rec.get("source_kind", "")) if url else "") + dt = rec.get("data_type") if dt in _NEAR_DUP_TYPES: rec.setdefault("focus", "") rec.setdefault("source_chars", 0) rec.setdefault("verified", "legacy") rec.setdefault("verifier_score", 0) if dt == "preference_pair": - rec.setdefault("chosen_chars", len(rec.get("chosen", "") or "")) - rec.setdefault("rejected_chars", len(rec.get("rejected", "") or "")) - rec["rejected"] = _strip_spurious_placeholder(rec.get("rejected", "") or "") or rec.get("rejected", "") + rec.setdefault("chosen_chars", len(rec.get("chosen") or "")) + rec.setdefault("rejected_chars", len(rec.get("rejected") or "")) + rec["rejected"] = _strip_spurious_placeholder(rec.get("rejected") or "") or rec.get("rejected") if dt == "tool_use_trace": ok = rec.get("outcome") == "success" rec.setdefault("trainable", ok) rec.setdefault("failure_reason", "" if ok else (rec.get("outcome") or "unknown")) rec.setdefault("n_error_results", _count_error_results(rec.get("steps"))) rec.setdefault("reasoning_spill", _has_reasoning_spill(rec.get("steps"))) + rec.setdefault("split", split_for_record(rec)) rec["schema_version"] = SCHEMA_VERSION + _normalize_envelope_fields(rec) return json.dumps(rec, sort_keys=True, ensure_ascii=False) != before def migrate_existing_records(): - changed_files = 0 - for dt in DATA_TYPES: - for split in SPLITS: - path = os.path.join(type_dir(dt), f"{split}.jsonl") - rows = list(_read_jsonl(path)) - if not rows: - continue - changed = False + """v10: normalizes THIS worker's own shard only -- one physical file, not a DATA_TYPES x SPLITS + folder tree.""" + rows = _load_shard_records() + if not rows: + return + changed = False + for rec in rows: + changed = _backfill(rec) or changed + if changed: + tmp = SHARD_PATH + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: for rec in rows: - changed = _backfill(dt, rec) or changed - if changed: - tmp = path + ".tmp" - with open(tmp, "w", encoding="utf-8") as f: - for rec in rows: - f.write(json.dumps(rec, ensure_ascii=False) + "\n") - os.replace(tmp, path) - changed_files += 1 - if changed_files: - print(f"[migrate] back-filled v{SCHEMA_VERSION} columns in {changed_files} file(s)", flush=True) - - -def build_audit() -> dict: + f.write(json.dumps(rec, ensure_ascii=False) + "\n") + os.replace(tmp, SHARD_PATH) + print(f"[migrate] back-filled v{SCHEMA_VERSION} columns in {SHARD_PATH}", flush=True) + + +def build_audit(records: list = None) -> dict: """Numbers that let you judge the data BEFORE training on it (shortcut features, coverage, verification, - split leakage). Written into MANIFEST.json and shown in the README.""" + split leakage). Written into MANIFEST.json and shown in the README. v10: works off an in-memory record + list (this worker's shard, or the full cross-worker merge) instead of re-reading per-type/per-split files.""" + records = _load_shard_records() if records is None else records audit = {} - pairs = [r for sp in SPLITS for r in _read_jsonl(os.path.join(type_dir("preference_pair"), f"{sp}.jsonl"))] + pairs = [r for r in records if r.get("data_type") == "preference_pair"] if pairs: - ratios = sorted(len(r.get("chosen", "")) / max(1, len(r.get("rejected", ""))) for r in pairs) + ratios = sorted(len(r.get("chosen") or "") / max(1, len(r.get("rejected") or "")) for r in pairs) audit["preference_pair"] = { "n": len(pairs), - "chosen_longer_pct": round(100 * sum(1 for r in pairs if len(r.get("chosen", "")) > len(r.get("rejected", ""))) / len(pairs), 1), + "chosen_longer_pct": round(100 * sum(1 for r in pairs if len(r.get("chosen") or "") > len(r.get("rejected") or "")) / len(pairs), 1), "median_len_ratio_chosen_over_rejected": round(ratios[len(ratios) // 2], 2), - "rejected_with_placeholder_pct": round(100 * sum(1 for r in pairs if _PLACEHOLDER_RE.search(r.get("rejected", "") or "")) / len(pairs), 1), + "rejected_with_placeholder_pct": round(100 * sum(1 for r in pairs if _PLACEHOLDER_RE.search(r.get("rejected") or "")) / len(pairs), 1), "note": "if chosen_longer_pct is near 100 or the placeholder share is high, a classifier can win on length / " "markers instead of judgment -- rebalance before DPO."} taxonomies = {("preference_pair", "category"): CATEGORY_TAXONOMY, ("sft_chat", "domain"): DOMAIN_TAXONOMY, ("reasoning_qa", "domain"): REASONING_DOMAINS} gaps = {} for (dt, field_name), tax in taxonomies.items(): - present = {r.get(field_name) for sp in SPLITS for r in _read_jsonl(os.path.join(type_dir(dt), f"{sp}.jsonl"))} + present = {r.get(field_name) for r in records if r.get("data_type") == dt} gaps[f"{dt}.{field_name}"] = sorted(set(tax) - present) audit["coverage_gaps"] = gaps verified, per_source = {}, {} for dt in _NEAR_DUP_TYPES: c = Counter() - for sp in SPLITS: - for r in _read_jsonl(os.path.join(type_dir(dt), f"{sp}.jsonl")): - c[r.get("verified", "legacy")] += 1 - if r.get("source_url"): - per_source.setdefault(_norm_source_key(r["source_url"]), set()).add(sp) + for r in records: + if r.get("data_type") != dt: + continue + c[r.get("verified", "legacy")] += 1 + if r.get("source_url"): + per_source.setdefault(_norm_source_key(r["source_url"]), set()).add(r.get("split")) if c: verified[dt] = dict(c) audit["verification"] = verified @@ -3881,36 +4435,37 @@ def build_audit() -> dict: def build_readme(manifest: dict): + """v10: ONE physical file (MERGED_REPO_PATH) instead of one config per data_type -- every row already + carries the same top-level columns (see _normalize_envelope_fields), which is what lets the Hub's + dataset viewer infer a single stable schema over rows of every data_type mixed together. The YAML + `configs:` block needs at least one named split, so the whole physical file is declared as split + `train`; the in-row `split` column (train/validation/test) is how to actually partition it -- documented + below. Anonymized: no mention anywhere of which underlying model produced a row (see the v10 + anonymization note at append_record()) -- that information simply is not part of the public schema + anymore, it is not just hidden here.""" types = manifest["data_types"] - configs = [(dt, [(s, f"{i['folder']}/{s}.jsonl") for s in SPLITS if i["splits"].get(s, 0) > 0]) - for dt, i in types.items()] - configs = [(dt, files) for dt, files in configs if files] out = ["---", "language:", "- ar", "- en", "license: other", "pretty_name: gijl style dataset (multi-type)", "tags:", - "- synthetic", "- preference", "- sft", "- tool-use", "- reasoning"] - if configs: - out.append("configs:") - for i, (dt, files) in enumerate(configs): - out += [f"- config_name: {dt}", " data_files:"] - for s, p in files: - out += [f" - split: {s}", f" path: {p}"] - if i == 0: - out.append(" default: true") - out += ["---", "", "# gijl style dataset (multi-type)", "", - f"Generated by `{MODEL_ID}` through a tool-using scouting loop over real sources (Stack Exchange, GitHub, " - "OSV, Hacker News, arXiv, Wikipedia, web). **Synthetic, model-written, only partly machine-verified, not " - "human-verified.** Every `rejected` response is intentionally poor and must never be used as an " - "example of good behavior.", "", - "| config | folder | train | validation | test | what it is |", "|---|---|---|---|---|---|"] + "- synthetic", "- preference", "- sft", "- tool-use", "- reasoning", "configs:", + f"- config_name: default", " data_files:", f" - split: train", f" path: {MERGED_REPO_PATH}", + " default: true"] + out += ["---", "", "# gijl agentic-thinker dataset (multi-type, single file)", "", + "Generated by a tool-using scouting loop over real sources (Stack Exchange, GitHub, OSV, Hacker News, " + "arXiv, Wikipedia, web). **Synthetic, model-written, only partly machine-verified, not human-verified.** " + "Every `rejected` response is intentionally poor and must never be used as an example of good behavior.", "", + f"Everything lives in one file, `{MERGED_REPO_PATH}`. Use the `data_type` column to select a record kind " + "and the `split` column (`train` / `validation` / `test`) to partition it -- the physical Hub split " + "('train') is just this one file; `split` is the real, meaningful partition.", "", + "| data_type | train | validation | test | what it is |", "|---|---|---|---|---|"] for dt, i in types.items(): sp = i["splits"] - out.append(f"| `{dt}` | `{i['folder']}/` | {sp.get('train', 0)} | {sp.get('validation', 0)} | " + out.append(f"| `{dt}` | {sp.get('train', 0)} | {sp.get('validation', 0)} | " f"{sp.get('test', 0)} | {i['description']} |") out += ["", "## Splits", "", "The split is a deterministic hash of the **source URL** (90/5/5), so every record derived from one source -- " "in any data type -- lands in the same split (no source appears in train and test). Sources already present " "keep their split. With a few hundred records the validation/test splits are too small to measure anything: " "treat them as smoke tests until the dataset is larger, and keep a separate protected eval set.", "", - "## Quality fields (schema v3)", "", + f"## Quality fields (schema v{SCHEMA_VERSION})", "", "- `verified`: `pass` (a second review pass accepted it), `unavailable` (reviewer gave no usable verdict), " "`unverified` (verification was off), `legacy` (written before v5, never reviewed). `verifier_score`: 1-5.", "- `run_id`: the collection session (tranche) that wrote the row. `focus`: the planned taxonomy cell; compare it " @@ -3918,31 +4473,31 @@ def build_readme(manifest: dict): "- `source_license`: license hint of the source text. Several sources are CC-BY-SA; prompts are paraphrases of " "those posts, so attribution / share-alike may apply to what you publish or train on.", "- `source_chars`: length of the material the example was grounded in.", - "- `tool_use_trace` (v6): `role` is always `harness_scouting` and `trainable` is always **false** -- this " + "- `tool_use_trace`: `role` is always `harness_scouting` and `trainable` is always **false** -- this " "is the scout's own search for raw material, not a model doing a task. `failure_reason`, " "`n_error_results`, `reasoning_spill` (reasoning that leaked into visible text) describe that process.", - "- `agent_trajectory` (v6): `role` is `agent_task`; `trainable` is true only when the agent finished " - "cleanly (a clean `finish_task_answer` call that passed verification). This is the real tool-use data.", - "- `collector_tag` (v6): which of the (up to 3) concurrently-running model workers wrote the row -- " - "see `generated_by` for the exact model id, and the top-of-file docstring for why 3 workers exist.", "", + "- `agent_trajectory` (v10): a code-enforced think -> act -> propose -> verify -> (error-recovery) -> " + "review -> output trajectory (see `think`, `research_steps`, `attempts`, `review`, `output`). Each entry " + "in `attempts` records a real verification (`code_execution`, `grounding`, or `self_consistency`) run by " + "the harness, not a model self-rating; `attempts[].error_recovery` is only present after a failed " + "verification, and records the external sources searched and how the candidate fixes were reconciled. " + "`trainable_for` lists which aspect of the row is safe to train on (`final_answer`, `trajectory_discipline`, " + "or both) -- a row that exhausted its verification attempts can still be a good trajectory-discipline " + "example while not being a good final-answer example.", "", "## How to use", "", "- Train on `verified == \"pass\"` rows; treat `legacy` / `unavailable` as unreviewed.", "- Use `preference_pair` only after checking the `audit` block below for length / marker shortcuts.", - "- For agentic / tool-use fine-tuning, use `agent_trajectory` rows with `trainable == true` -- **never** " - "`tool_use_trace`, which is pipeline-internal bookkeeping and is never agent behavior (v6; earlier " - "README text incorrectly suggested filtering `tool_use_trace` by `trainable` for this purpose).", - f"- Teacher is `{MODEL_ID}` (worker `{MODEL_TAG}`; two sibling workers write the same schema under a " - "different `collector_tag`/`generated_by`). A student that shares a teacher's base model can inherit " - "that teacher's traits through generated data even when the data looks unrelated (Subliminal Learning, " - "2025); run a behavioral check on a held-out set, per teacher.", ""] + "- For agentic / tool-use fine-tuning, use `agent_trajectory` rows and check `trainable_for` -- **never** " + "`tool_use_trace`, which is pipeline-internal bookkeeping and is never agent behavior.", ""] audit = manifest.get("audit") or {} if audit: out += ["## Audit", "", "```json", json.dumps(audit, indent=2, ensure_ascii=False), "```", ""] - out += ["## Shared envelope", "", - "`id, schema_version, data_type, loop, language, content_hash, source_kind, source_url, generated_by, " - "generated_at, run_id, focus, source_license, source_chars, verified, verifier_score, collector_tag` " - "(v6: which concurrent model worker wrote the row) on every data record. " - "`content_hash` is used for exact de-duplication; a bigram-Jaccard check on the primary text catches near-duplicates.", "", + out += ["## Shared columns", "", + "`id, schema_version, data_type, loop, language, split, content_hash, source_kind, source_url, " + "generated_at, run_id, focus, source_license, source_chars, verified, verifier_score, trainable_for` " + "on every row, regardless of data_type -- this is what keeps one consistent schema across the whole " + "file. `content_hash` is used for exact de-duplication; a bigram-Jaccard check on the primary text " + "catches near-duplicates. Fields specific to one data_type are `null` on rows of every other type.", "", "## Loops", ""] for l in LOOPS.values(): out.append(f"- `{l.name}` (target weight {l.weight}): {l.goal}") @@ -3954,9 +4509,11 @@ def build_readme(manifest: dict): f.write("\n".join(out)) -def refresh_docs(): +def refresh_docs(records: list = None): + """Offline-safe local refresh (this worker's own shard only) when called with no records -- see + merge_shards_for_sync() in the sync section for the full cross-worker refresh done at sync time.""" try: - build_readme(build_manifest()) + build_readme(build_manifest(records)) except Exception as e: # noqa: BLE001 -- docs must never break a run print(f"[docs] manifest/README refresh failed: {_redact(str(e))}", flush=True) @@ -4076,19 +4633,24 @@ class TraceRecorder: "tool_results": [{"name": n, "content": _scrub_sensitive(_trunc(r, TRACE_RESULT_CHARS))} for n, _, r, _ in calls]}) def build(self, outcome: str, written_ids: list) -> dict: + # v10: this flat shape (steps/goal/tools_offered/...) is used for data_type="tool_use_trace" (the + # scout's own harness_scouting search) only now -- the real agent_task trajectory has its own, + # much richer shape built directly in run_agent_task()'s record assembly step, + # not through TraceRecorder, since it needed a genuinely different structure (think/research_steps/ + # attempts/review/output), not just a role label on the same fields. p = self.plan rid = _rid("trace", self.data_type, str(p.cycle), p.loop.name, str(self.t0)) ok = outcome == "success" - trainable = False if self.role == "harness_scouting" else ok return {"id": rid, "schema_version": SCHEMA_VERSION, "data_type": self.data_type, "loop": p.loop.name, "language": p.language, "content_hash": _content_hash(rid), "source_kind": "", "source_url": "", - "generated_by": MODEL_ID, "generated_at": _now(), "collector_tag": MODEL_TAG, "role": self.role, + "generated_at": _now(), "role": self.role, "cycle": p.cycle, "focus": p.focus_label, "goal": self.goal, "tools_offered": self.tools, "steps": self.steps, "num_tool_calls": sum(len(s["tool_calls"]) for s in self.steps), "outcome": outcome, "records_written": written_ids, "origin_mix": "+".join(sorted(self.origins)) or "none", "elapsed_s": round(time.time() - self.t0, 1), - "run_id": RUN_ID, "source_license": "", "trainable": trainable, "failure_reason": "" if ok else outcome, + "run_id": RUN_ID, "source_license": "", "trainable": False, "trainable_for": None, + "failure_reason": "" if ok else outcome, "n_error_results": _count_error_results(self.steps), "reasoning_spill": _has_reasoning_spill(self.steps)} @@ -4099,11 +4661,11 @@ def make_source_record(plan: CyclePlan, identifier: str, scenario: dict, status: rid = _rid("src", plan.loop.name, url) rec = {"id": rid, "schema_version": SCHEMA_VERSION, "data_type": "source_index", "loop": plan.loop.name, "language": plan.language, "content_hash": _content_hash(url), "source_kind": cand.get("kind", ""), - "source_url": url, "generated_by": MODEL_ID, "generated_at": _now(), + "source_url": url, "generated_at": _now(), "title": _scrub_sensitive(cand.get("title", ""))[:300], "chars": len(body), "body_sha256": hashlib.sha256(body.encode("utf-8")).hexdigest() if body else "", "status": status, "error": (error or "")[:300], "focus": plan.focus_label, - "run_id": RUN_ID, "source_license": _license_hint(url, cand.get("kind", "")), "collector_tag": MODEL_TAG} + "run_id": RUN_ID, "source_license": _license_hint(url, cand.get("kind", "")), "trainable_for": None} if STORE_SOURCE_BODIES: rec["body_excerpt"] = body[:MAX_PROMPT_CHARS] return rec @@ -4352,12 +4914,16 @@ def run_cycle(client, state: RunState, cycle: int, deadline: float): # --------------------------------------------------------------------------- -# 12b. v7, kimi-only: cross-review pass. Addresses the self-review limitation stated at _VERIFY_HEAD ("the +# 12b. v10, kimi-only: cross-review pass. Addresses the self-review limitation stated at _VERIFY_HEAD ("the # reviewer is the same model that wrote the record") with no change to the shared verify_generated()/ -# _VERIFY_TEMPLATES machinery used by every worker's OWN records: this is a second, separate check that Kimi runs -# on OTHER workers' already-written, already-synced rows, using the SAME per-loop verifier templates (imported -# unchanged) against a DIFFERENT model. It never edits the row being reviewed -- it only ever appends a new -# "cross_review" record of its own, so it cannot race with the worker that owns that row. +# _VERIFY_TEMPLATES machinery used by every worker's OWN records: this is a second, separate check this worker +# runs on OTHER workers' already-written, already-synced rows, using the SAME per-loop verifier templates +# (imported unchanged) against a DIFFERENT model. It never edits the row being reviewed -- it only ever appends a +# new "cross_review" record of its own, so it cannot race with the worker that owns that row. v10: re-pointed at +# the one-shard-per-worker layout (_shard_id_for(), see WHAT CHANGED IN v10) instead of the old per-type/per-split +# folders, updated to the flat agent_trajectory envelope (research_steps/attempts/output, not steps/final_answer), +# and the written record no longer stores which worker wrote the row it reviewed (reviewed_id alone -- a sha256 +# hash, see _rid() -- already uniquely identifies it) to match this project's anonymization requirement. # --------------------------------------------------------------------------- CROSS_REVIEW_EVERY_N_CYCLES = int(os.environ.get("GIJL_CROSS_REVIEW_EVERY_N", "5")) CROSS_REVIEW_BATCH = int(os.environ.get("GIJL_CROSS_REVIEW_BATCH", "3")) @@ -4380,43 +4946,52 @@ def _save_cross_reviewed(seen: dict): def _fetch_other_worker_candidates(tag: str, limit: int) -> list: - """Pull a few of `tag`'s own already-synced agent_trajectory/sft_chat rows that are not verified=='pass', - straight from the hub (read-only -- this process never writes into another worker's folder). Mirrors - restore_from_hf()'s own snapshot_download() call, just narrowed to one other worker's two loops.""" + """Pull `tag`'s own already-synced shard (v10: one physical file per worker, located the same + deterministic, non-revealing way _shard_id_for() computes this worker's own -- see WHAT CHANGED IN v10) + and keep its agent_trajectory/sft_chat rows that are not verified=='pass'. Read-only: this process never + writes into another worker's shard. Mirrors restore_from_hf()'s own snapshot_download() call, just + pointed at one other worker's shard instead of this worker's own.""" token = os.environ.get("HF_TOKEN") if not token: return [] from huggingface_hub import snapshot_download + other_shard_id = _shard_id_for(tag) + other_shard_rel = f"data/_shards/{other_shard_id}.jsonl" local = os.path.join(WORK_DIR, "_cross_review_src", tag) try: _with_retry(lambda: snapshot_download( repo_id=TARGET_DATASET_REPO, repo_type="dataset", token=token, local_dir=local, - allow_patterns=[f"agent_trajectory_dataset/{tag}/*.jsonl", f"sft_dataset/{tag}/*.jsonl"]), + allow_patterns=[other_shard_rel]), f"cross_review_pull[{tag}]") except Exception as e: # noqa: BLE001 -- that worker may not have synced anything yet print(f"[cross_review] nothing to pull yet from {tag} ({_redact(str(e))[:120]})", flush=True) return [] out = [] - for dt, folder in (("agent_trajectory", "agent_trajectory_dataset"), ("sft_chat", "sft_dataset")): - for split in SPLITS: - for rec in _read_jsonl(os.path.join(local, folder, tag, f"{split}.jsonl")): - if rec.get("verified") != "pass" and rec.get("trainable", True): - out.append((dt, rec)) + for rec in _read_jsonl(os.path.join(local, other_shard_rel)): + dt = rec.get("data_type") + if dt in ("agent_trajectory", "sft_chat") and rec.get("verified") != "pass" and rec.get("trainable", True): + out.append((dt, rec)) random.Random(f"cross-review-{tag}").shuffle(out) return out[:limit] def _cross_review_verify_input(dt: str, rec: dict): """Reuses the EXISTING per-loop verifier templates (agent_task / sft_helpful) unchanged -- only the model - doing the judging is different. Shaped like _verify_input() but takes a raw record instead of a live plan.""" + doing the judging is different. Shaped like _verify_input() but takes a raw record instead of a live plan. + v10: agent_trajectory rows are the flat envelope's research_steps/attempts/output shape (see WHAT CHANGED + IN v10), not the old steps/final_answer shape.""" if dt == "agent_trajectory": lines = [] - for i, s in enumerate(rec.get("steps", []), 1): - for c, r in zip(s.get("tool_calls", []), s.get("tool_results", [])): - lines.append(f"{i}. {c['name']}({c['arguments_json']}) -> {r['content'][:400]}") + for i, s in enumerate(rec.get("research_steps") or [], 1): + lines.append(f"{i}. {s.get('tool','')}({s.get('arguments_json','')}) -> {str(s.get('result',''))[:400]}") + for a in (rec.get("attempts") or []): + v = a.get("verification") or {} + lines.append(f"[verify attempt {a.get('attempt','?')}] method={v.get('method','?')} passed={v.get('passed')}") steps_text = "\n".join(lines) or "(no tool calls)" + output = rec.get("output") or {} + final_text = output.get("content", "") if isinstance(output, dict) else str(output or "") return (f"\n{rec.get('task_prompt','')}\n\n\n\n{steps_text}\n\n\n" - f"\n{rec.get('final_answer','')}\n"), "agent_task" + f"\n{final_text}\n"), "agent_task" m = rec.get("messages") or [{}, {}] prompt = m[0].get("content", "") if m else "" response = m[1].get("content", "") if len(m) > 1 else "" @@ -4457,11 +5032,14 @@ def cross_review_pass(client, state: RunState): continue score = int(float(parsed.get("score", 0) or 0)) verdict = str(parsed.get("verdict") or parsed.get("better") or "").strip().lower() + # v10: no generated_by/collector_tag (removed from the schema entirely) and no reviewed_tag -- + # reviewed_id (a sha256 hash, see _rid()) is already a unique join key on its own, so this row + # never stores which worker wrote the row it reviewed; see WHAT CHANGED IN v10. out = {"id": _rid("cross_review", tag, dt, rec.get("id", "")), "schema_version": SCHEMA_VERSION, "data_type": "cross_review", "loop": loop_name, "language": rec.get("language", ""), "content_hash": _content_hash(content), "source_kind": "", "source_url": "", - "generated_by": MODEL_ID, "generated_at": _now(), "collector_tag": MODEL_TAG, - "reviewed_tag": tag, "reviewed_id": rec.get("id", ""), "reviewed_type": dt, + "generated_at": _now(), + "reviewed_id": rec.get("id", ""), "reviewed_type": dt, "reviewed_verified": rec.get("verified", ""), "verdict": verdict, "score": score, "issues": [str(i)[:150] for i in (parsed.get("issues") or [])][:4], "run_id": RUN_ID, "source_license": "", "focus": "", "source_chars": 0, @@ -4482,15 +5060,55 @@ def cross_review_pass(client, state: RunState): # 13. Sync (all data folders + state + manifest + README) to the hub # --------------------------------------------------------------------------- def _sync_patterns() -> list: - # v6: every data folder is ALREADY namespaced under MODEL_TAG by type_dir() (e.g. "dpo_dataset/glm/*"), and - # fnmatch (what huggingface_hub's allow_patterns uses) matches "/" with "*" same as any other character, so - # "dpo_dataset/*" still matches "dpo_dataset/glm/train.jsonl" unchanged. Only the two UN-namespaced literal - # filenames (MANIFEST.json/README.md) needed a MODEL_TAG suffix -- done where they are defined above. - pats = ([f"{m['folder']}/*" for m in DATA_TYPES.values()] - + [f"state/{MODEL_TAG}/*", os.path.basename(MANIFEST_PATH), os.path.basename(README_PATH)]) + # v10: ONE shard file per worker (SHARD_PATH, race-free -- single writer, see append_record()) plus the + # derived MERGED_PATH that merge_shards_for_sync() rebuilds fresh on every sync. MANIFEST.json/README.md + # are un-namespaced now (global, built from the merge) instead of per-MODEL_TAG files. + pats = [f"data/_shards/{SHARD_ID}.jsonl", MERGED_REPO_PATH, f"state/{MODEL_TAG}/*", + os.path.basename(MANIFEST_PATH), os.path.basename(README_PATH)] return pats + (["bin/*"] if CACHE_LLAMA_BINARY and BACKEND == "llamacpp" else []) +def merge_shards_for_sync(token: str) -> list: + """v10: pure, derived step run on every sync -- download every worker's CURRENT shard from the hub, + concatenate with this worker's own (already up to date on local disk), de-duplicate by id, and sort + deterministically so the result is reproducible no matter which worker performs the merge or in what + order. This is what makes it safe for more than one of the 5 workers to regenerate and upload + MERGED_PATH: if two merges race on the hub, the one that loses the commit race is at worst missing a + few of the other worker's newest rows -- never corrupted, never duplicated -- and that gap closes at + the very next sync (every SYNC_INTERVAL_S), because each merge always starts over from the + authoritative, single-writer shard files, never from a previous merge's output.""" + records = {rec.get("id"): rec for rec in _load_shard_records() if rec.get("id")} + try: + from huggingface_hub import HfApi, snapshot_download + api = HfApi(token=token) + try: + files = api.list_repo_files(repo_id=TARGET_DATASET_REPO, repo_type="dataset") + except Exception: + files = [] + other_shards = [p for p in files if p.startswith("data/_shards/") and p.endswith(".jsonl") + and p != f"data/_shards/{SHARD_ID}.jsonl"] + if other_shards: + _with_retry(lambda: snapshot_download( + repo_id=TARGET_DATASET_REPO, repo_type="dataset", token=token, local_dir=WORK_DIR, + allow_patterns=other_shards), "merge-download") + for p in other_shards: + for rec in _read_jsonl(os.path.join(WORK_DIR, p)): + rid = rec.get("id") + if rid and rid not in records: # this worker's own copy of a row always wins a tie + records[rid] = rec + except Exception as e: + print(f"[merge] could not fetch other workers' shards this round ({_redact(str(e))}); " + f"uploading with just this worker's own data for now", flush=True) + merged = [records[k] for k in sorted(records)] + os.makedirs(os.path.dirname(MERGED_PATH), exist_ok=True) + tmp = MERGED_PATH + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + for rec in merged: + f.write(json.dumps(rec, ensure_ascii=False) + "\n") + os.replace(tmp, MERGED_PATH) + return merged + + def _with_retry(fn, what: str, tries: int = 3, base_delay: float = 2.0): """v6: three Spaces now push to the SAME TARGET_DATASET_REPO concurrently. Every path this process writes is disjoint (MODEL_TAG namespacing, above) -- that is the real fix for the race -- but the hub still serializes @@ -4510,40 +5128,41 @@ def _with_retry(fn, what: str, tries: int = 3, base_delay: float = 2.0): def restore_from_hf(): - """Pull the data folders + state back before starting -- without this a - fresh Kaggle container has no memory of what earlier sessions picked.""" + """Pull this worker's own shard + state back before starting -- without this a fresh container has + no memory of what earlier sessions picked. v10: only the one shard file is needed (not the other + workers' shards, and not MERGED_PATH, which merge_shards_for_sync() rebuilds from scratch every + sync anyway).""" token = os.environ.get("HF_TOKEN") if not token: return try: from huggingface_hub import snapshot_download - # v6: only THIS worker's own shard is needed for resumability (state/, dpo_dataset/, ...) -- - # fnmatch "*" matches "/" (see _sync_patterns), so narrowing to f"*/{MODEL_TAG}/*" avoids pulling the - # other two workers' data down into every container unnecessarily. _with_retry(lambda: snapshot_download( repo_id=TARGET_DATASET_REPO, repo_type="dataset", token=token, local_dir=WORK_DIR, - allow_patterns=[f"{m['folder']}/{MODEL_TAG}/*" for m in DATA_TYPES.values()] + [f"state/{MODEL_TAG}/*"] + allow_patterns=[f"data/_shards/{SHARD_ID}.jsonl", f"state/{MODEL_TAG}/*"] + (["bin/*"] if CACHE_LLAMA_BINARY and BACKEND == "llamacpp" else [])), "restore") - print(f"[restore] pulled prior state from {TARGET_DATASET_REPO} (worker={MODEL_TAG})", flush=True) + print(f"[restore] pulled prior state from {TARGET_DATASET_REPO}", flush=True) except Exception as e: print(f"[restore] nothing to restore yet ({_redact(str(e))})", flush=True) def sync_to_hf(): - refresh_docs() # MANIFEST..json + README..md are refreshed even when there is no token + refresh_docs() # offline-safe local refresh (this worker's own shard only), even with no token token = os.environ.get("HF_TOKEN") if not token: return try: from huggingface_hub import HfApi, create_repo - # exist_ok=True makes this race-safe across the 3 workers too: whichever starts first actually creates - # TARGET_DATASET_REPO, the other two just find it already there. + # exist_ok=True makes this race-safe across the 5 workers too: whichever starts first actually creates + # TARGET_DATASET_REPO, the others just find it already there. _with_retry(lambda: create_repo(TARGET_DATASET_REPO, repo_type="dataset", token=token, exist_ok=True, private=True), "create_repo") + merged = merge_shards_for_sync(token) + refresh_docs(merged) # the real, global MANIFEST.json/README.md -- built from every worker's data _with_retry(lambda: HfApi(token=token).upload_folder( repo_id=TARGET_DATASET_REPO, repo_type="dataset", folder_path=WORK_DIR, allow_patterns=_sync_patterns(), ignore_patterns=["*.tmp"]), "sync") - print(f"[sync] pushed to {TARGET_DATASET_REPO} (worker={MODEL_TAG})", flush=True) + print(f"[sync] pushed to {TARGET_DATASET_REPO} ({len(merged)} merged rows across all workers)", flush=True) except Exception as e: print(f"[sync] failed: {_redact(str(e))}", flush=True)