commit 044bd03f0edc8686f947adee911449cae64b3649 Author: Tom Lu Date: Thu Aug 6 22:53:47 2026 +0800 Initial Open-SWE-Traces cleanup pipeline diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..78d75d5 --- /dev/null +++ b/.env.example @@ -0,0 +1,15 @@ +# The API key must be provided at runtime and must never be committed. +GLM_API_KEY=replace-with-a-runtime-secret + +# The gateway exposes an OpenAI-compatible chat-completions route. +GLM_API_BASE=https://llm-api.cowin.run +GLM_API_PATH=/v1/chat/completions +GLM_MODEL=glm-5.2 + +# Conservative defaults for structured quality-control output. +GLM_TIMEOUT_SECONDS=300 +GLM_MAX_RETRIES=3 +GLM_MAX_TOKENS=8192 +GLM_TEMPERATURE=0.0 +GLM_REASONING_EFFORT=high +GLM_THINKING_ENABLED=true diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5bbdab0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,27 @@ +# Python environments and caches. +.venv/ +__pycache__/ +*.py[cod] +.pytest_cache/ +.ruff_cache/ +*.egg-info/ +build/ +dist/ + +# Local credentials. Never commit API keys. +.env +.env.* +!.env.example + +# Downloaded datasets and generated pipeline output. +raw/ +qc_outputs/ +outputs/ +samples/ +artifacts/ +*.log + +# Editor and operating-system files. +.DS_Store +.idea/ +.vscode/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..0e27093 --- /dev/null +++ b/README.md @@ -0,0 +1,325 @@ +# SWE Data Processing + +`swe-data-processing` is a conservative, auditable Python pipeline for cleaning +[`nvidia/Open-SWE-Traces`](https://huggingface.co/datasets/nvidia/Open-SWE-Traces) +before supervised fine-tuning of smaller coding agents. + +The project is designed for a restricted environment in which repository +containers and agent sandboxes are unavailable. It combines deterministic local +checks with GLM-5.2 API judgments. It does **not** claim that static inspection can +prove code correctness. Instead, it separates internally consistent silver +positives, useful negatives, safely repairable formatting issues, unverified +records, and corrupted records. + +## Dataset snapshot + +The current local snapshot contains 207,489 trajectories over 22,320 unique +issues. The immutable dataset outcome field is named `resolved`: + +| `resolved` | Meaning | Count | Percentage | +|---:|---|---:|---:| +| `1` | Successful candidate | 65,244 | 31.44% | +| `0` | Explicit failure | 95,487 | 46.02% | +| `-1` | Unknown outcome | 46,758 | 22.54% | + +Records must be split by `instance_id`, not by trajectory, to prevent the same +issue from leaking across train and evaluation sets. + +## Safety model + +Static cleanup may improve representation quality, but it may not create new +execution facts. The implementation enforces these invariants locally after +every model response: + +1. `resolved` is immutable. +2. A `resolved=0` record can never become a full successful SFT example. +3. A `resolved=-1` record cannot become `SFT_FULL` without execution. +4. Existing tool observations cannot be rewritten or fabricated. +5. Model and reference patches are immutable during static repair. +6. Truncated records have a maximum use of `SFT_STEP_ONLY`. +7. A silver positive requires reliable post-edit verification, no hard failures, + and a passing value for every QC dimension. +8. Every repair requires a separate review API call. + +The API model proposes decisions and repair plans. Deterministic Python code +validates schemas, enforces outcome policy, applies only allowlisted edits, and +records provenance. + +Long trajectories are sent to the API through a turn-preserving evidence view. +Test observations, state-changing turns, malformed calls, final turns, patches, +and stable turn IDs receive priority. Shortened values include their original +character count and SHA-256 hash. This compaction affects only the API prompt; +classification never rewrites source JSONL or Parquet records. + +## API endpoint + +The configured gateway is OpenAI chat-completions compatible: + +```text +POST https://llm-api.cowin.run/v1/chat/completions +``` + +`/v1/text-completion` and `/v1/text-completions` resolve to the gateway's web +console rather than an inference API, so they are not used. + +The default model is `glm-5.2`. The API key is read only from `GLM_API_KEY`. Do +not write a key into source code, command history, output manifests, or this +README. + +## Project layout + +```text +swe_data_processing/ +├── pyproject.toml +├── README.md +├── .env.example +├── src/swe_data_processing/ +│ ├── cli.py # Command-line entry points +│ ├── client.py # GLM API client and JSON validation +│ ├── config.py # Environment-only runtime settings +│ ├── evidence.py # Prompt-only evidence compaction +│ ├── features.py # Deterministic static evidence extraction +│ ├── io.py # Streaming Parquet and JSONL readers +│ ├── policy.py # Immutable local policy guards +│ ├── repair.py # Allowlisted deterministic repair application +│ ├── resources.py # Packaged prompt/schema loading +│ ├── workflow.py # Classify, plan repair, and review stages +│ ├── prompts/ # Version-controlled GLM system prompts +│ └── schemas/ # JSON Schemas for every API stage +├── tests/ # Offline unit tests; no network calls +├── scripts/ # Existing sampling and profiling utilities +├── raw/Open-SWE-Traces/ # Downloaded dataset; ignored by Git +├── samples/ # Human-review samples +├── reports/ # Dataset reports and QC rubrics +└── qc_outputs/ # Generated manifests; ignored by Git +``` + +## Installation + +The existing remote virtual environment can install the package in editable +mode: + +```bash +cd /mnt/beegfs/yi/swe_data_processing +./.venv/bin/pip install -e '.[dev]' +``` + +For a fresh environment: + +```bash +python3 -m venv .venv +./.venv/bin/pip install --upgrade pip +./.venv/bin/pip install -e '.[dev]' +``` + +## Configuration + +Export credentials in the shell that launches the pipeline: + +```bash +export GLM_API_KEY='your-runtime-secret' +export GLM_API_BASE='https://llm-api.cowin.run' +export GLM_API_PATH='/v1/chat/completions' +export GLM_MODEL='glm-5.2' +``` + +Optional settings and their defaults: + +```bash +export GLM_TIMEOUT_SECONDS=300 +export GLM_MAX_RETRIES=3 +export GLM_MAX_TOKENS=8192 +export GLM_TEMPERATURE=0.0 +export GLM_REASONING_EFFORT=high +export GLM_THINKING_ENABLED=true +``` + +If the gateway rejects GLM-specific `thinking` or `reasoning_effort` fields, the +client automatically retries with the portable OpenAI-compatible request subset. + +## Commands + +### 1. Verify API authentication and structured output + +```bash +swe-qc smoke-test +``` + +The command prints the endpoint, model, request ID, usage, and a tiny validated +JSON response. It never prints the API key. + +### 2. Extract deterministic evidence without API calls + +From the 20-record review sample: + +```bash +mkdir -p qc_outputs +swe-qc features \ + --input samples/sample_20_seed_20260805.jsonl \ + --output qc_outputs/sample20.features.jsonl \ + --errors qc_outputs/sample20.features.errors.jsonl +``` + +From all Parquet shards: + +```bash +swe-qc features \ + --input raw/Open-SWE-Traces \ + --output qc_outputs/all.features.jsonl \ + --errors qc_outputs/all.features.errors.jsonl \ + --resume +``` + +Features include malformed tool arguments, unknown tools, role alternation, +state-changing turns, post-edit test evidence, masked shell pipelines, patch file +sets, patch size ratios, and explicit user constraints. + +### 3. Classify trajectories through GLM-5.2 + +Run a small pilot first: + +```bash +swe-qc classify \ + --input samples/sample_20_seed_20260805.jsonl \ + --output qc_outputs/sample20.classifications.jsonl \ + --errors qc_outputs/sample20.classification.errors.jsonl \ + --limit 20 \ + --resume +``` + +The classifier returns one of: + +- `ACCEPT_SILVER_POSITIVE` +- `ACCEPT_NEGATIVE` +- `STATIC_REPAIR` +- `HOLD_UNVERIFIED` +- `REJECT` + +Training use is tracked separately as `SFT_FULL`, `SFT_STEP_ONLY`, +`DPO_REJECTED`, `ERROR_ANALYSIS`, `HOLD`, or `DROP`. + +### 4. Audit educational process quality + +This stage does not repair trajectories. It identifies defensible erroneous or +inefficient calls, scores six process-quality dimensions, and optionally finds +a causal first-bad assistant turn for prefix-only learning: + +```bash +swe-qc audit \ + --input samples/sample_20_seed_20260805.jsonl \ + --output qc_outputs/sample20.audits.jsonl \ + --errors qc_outputs/sample20.audit.errors.jsonl \ + --resume +``` + +The deterministic score combines weighted process dimensions with penalties +for minor, major, and critical behavior issues. Failed exploratory calls are +not penalized when the agent interprets them correctly and recovers. + +The causal audit payload excludes the reference patch and every derived signal, +including reference file names, patch length, and model/reference size ratios. +The immutable outcome is used only to select the mode: failed or unknown +trajectories can become process-prefix candidates, but never full-trajectory +SFT candidates. + +### 5. Create static repair plans + +Only classifications that explicitly return `STATIC_REPAIR` are processed by +default: + +```bash +swe-qc repair-plan \ + --input samples/sample_20_seed_20260805.jsonl \ + --classifications qc_outputs/sample20.classifications.jsonl \ + --output qc_outputs/sample20.repair-plans.jsonl \ + --errors qc_outputs/sample20.repair-plan.errors.jsonl \ + --resume +``` + +The model may propose only operations from the allowlist. Code changes, test +result synthesis, patch replacement, and outcome upgrades are prohibited. + +### 6. Apply plans deterministically + +```bash +swe-qc apply-repair \ + --input samples/sample_20_seed_20260805.jsonl \ + --plans qc_outputs/sample20.repair-plans.jsonl \ + --output qc_outputs/sample20.repaired.jsonl \ + --errors qc_outputs/sample20.apply.errors.jsonl \ + --resume +``` + +The applier checks target turns, read-only pair deletion, immutable patch fields, +and retained tool-output hashes. It emits a structured diff with input and output +SHA-256 hashes. + +### 7. Independently review repairs + +```bash +swe-qc review \ + --input samples/sample_20_seed_20260805.jsonl \ + --classifications qc_outputs/sample20.classifications.jsonl \ + --plans qc_outputs/sample20.repair-plans.jsonl \ + --repaired qc_outputs/sample20.repaired.jsonl \ + --output qc_outputs/sample20.reviews.jsonl \ + --errors qc_outputs/sample20.review.errors.jsonl \ + --resume +``` + +Approval means that a static repair preserved evidence and structure. It does not +mean that the code patch was executed or proved correct. + +## Output and resume behavior + +Pipeline manifests are append-only JSONL. `--resume` reads completed sample IDs +from the output and skips them. Each API response includes non-sensitive +provenance: + +- model and endpoint; +- gateway request ID; +- token usage when available; +- input SHA-256; +- compatibility-fallback flag; +- UTC creation time. + +Errors are written as compact records containing sample ID, stage, exception +type, and a bounded message. Raw prompts and credentials are not copied into +error logs. + +## Recommended rollout + +1. Run offline tests. +2. Run `smoke-test` once. +3. Run classification on the 20 manually reviewed records. +4. Compare GLM decisions with the human labels. +5. Build a 2,000-record stratified calibration set. +6. Require at least 95% precision for `ACCEPT_SILVER_POSITIVE` before scaling. +7. Keep `resolved=0` as negative data rather than attempting to turn it into + positive trajectories. +8. Keep `resolved=-1` in an unverified manifest unless it contains explicit, + reliable failure evidence. + +## Development + +Run all offline tests: + +```bash +./.venv/bin/pytest +``` + +Run lint checks: + +```bash +./.venv/bin/ruff check src tests +``` + +Tests use an `httpx.MockTransport` and never contact the GLM endpoint. + +## Credential handling + +- `.env` files are ignored by Git. +- The provided API key is not stored anywhere in this project. +- Use a secret manager or a protected runtime environment variable for batch + jobs. +- Rotate the key if it has been copied into a public log, issue, or repository. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..957be2b --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,52 @@ +[build-system] +requires = ["setuptools>=70", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "swe-data-processing" +version = "0.1.0" +description = "Static quality control and repair planning for Open-SWE-Traces." +readme = "README.md" +requires-python = ">=3.10" +license = {text = "MIT"} +authors = [ + {name = "TIGER Lab"} +] +dependencies = [ + "httpx>=0.27,<1", + "jsonschema>=4.23,<5", + "pyarrow>=16,<26" +] + +[project.optional-dependencies] +dev = [ + "pytest>=8,<10", + "ruff>=0.9,<1" +] + +[project.scripts] +swe-qc = "swe_data_processing.cli:main" + +[tool.setuptools] +package-dir = {"" = "src"} +include-package-data = true + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.setuptools.package-data] +swe_data_processing = [ + "prompts/*.md", + "schemas/*.json" +] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-q" + +[tool.ruff] +line-length = 110 +target-version = "py310" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B"] diff --git a/reports/glm52_sample20_qc_audit_20260805.md b/reports/glm52_sample20_qc_audit_20260805.md new file mode 100644 index 0000000..5d454ba --- /dev/null +++ b/reports/glm52_sample20_qc_audit_20260805.md @@ -0,0 +1,179 @@ +# GLM-5.2 Sample-20 Classification and Repair Audit + +Date: 2026-08-05 +Dataset: `nvidia/Open-SWE-Traces` +Sample: uniform random sample of 20 trajectories, seed `20260805` + +## Executive result + +GLM-5.2 classified all 20 records. The direct full-SFT decision exactly matched +the existing manual review: samples 1, 12, 13, 15, 17, and 18 were accepted, +for a total of 6/20. There were no false rejects relative to the manual +keep-candidate set and no additional false-positive keeps. + +Truncation cannot turn any of the other 14 records into a complete successful +trajectory because it cannot create missing task-resolution evidence. The +count of additional `SFT_FULL` records obtainable by static truncation is +therefore 0/20. + +For step-level SFT, the current GLM repair planner has poor recall. Both the +original repair rubric and the revised explicit step-salvage rubric produced +zero `CREATE_STEP_EXAMPLE` proposals. A conservative human audit identified +four high-confidence step-only candidates and four borderline candidates. + +## Classification distribution + +| Classification | Count | +|---|---:| +| `ACCEPT_SILVER_POSITIVE / SFT_FULL` | 6 | +| `ACCEPT_NEGATIVE / DPO_REJECTED` | 7 | +| `HOLD_UNVERIFIED / HOLD` | 5 | +| `REJECT / DROP` | 2 | + +The source outcomes in the sample were six `resolved=1`, seven `resolved=0`, +and seven `resolved=-1`. All six `resolved=1` records were accepted and all +other records were prevented from becoming full-SFT positives. + +## Comparison with the prior manual review + +| Manual group | Samples | GLM result | +|---|---|---| +| Keep candidate | 1, 12, 13, 15, 17, 18 | All six accepted as `SFT_FULL` | +| Repair/replay | 7, 9, 10, 14 | Three held/rejected and one dropped; none repaired | +| Reject as positive | 2, 3, 4, 5, 6, 8, 11, 16, 19, 20 | None accepted as positive | + +The classification threshold is not too strict for full positive trajectories +on this sample: keep-set agreement is 20/20 and keep precision/recall against +the prior manual labels are both 6/6. The pipeline is, however, too strict at +recovering training value from repair/replay records. + +## Repair proposal v1 audit + +The first repair pass returned 20 schema-valid plans: + +| Repair decision | Count | +|---|---:| +| `NO_CHANGE` | 8 | +| `REQUIRES_EXECUTION` | 8 | +| `APPLY_STATIC_REPAIR` | 2 | +| `DROP` | 2 | + +It proposed four `REWRITE_FINAL_SUMMARY` operations and no truncations. + +The proposal set was not reliably executable: + +1. Sample 3 had a syntactically executable summary rewrite, but it would alter + an authentic rejected response. It adds no positive-SFT value and weakens + the original DPO negative signal. +2. Sample 4 used an object as the replacement for `REWRITE_FINAL_SUMMARY`, + while the deterministic applier requires a string. Application failed with + `PolicyViolation`. +3. Samples 6 and 14 used `REQUIRES_EXECUTION` while also including mutation + operations. The applier does not authorize mutation for that decision. + +Only one of the two `APPLY_STATIC_REPAIR` plans executed, and that successful +mutation remained `DPO_REJECTED`. Consequently v1 produced zero additional SFT +records. + +## Repair proposal v2 audit + +The rubric was revised to define strict step salvage and local policy was +strengthened to require: + +- no operations for `NO_CHANGE`, `REQUIRES_EXECUTION`, or `DROP`; +- no rewriting of authentic `DPO_REJECTED` trajectories; +- a complete assistant replacement plus truncation for every step example; +- operation-specific replacement types; +- `SFT_STEP_ONLY` as the maximum use for truncated examples. + +Seventeen policy tests plus Ruff checks passed after the change. + +GLM returned 18 policy-valid v2 plans and two locally rejected plans. Samples 3 +and 8 again tried to rewrite authentic DPO negatives and were rejected by the +new deterministic guard. The 18 valid plans contained 11 `NO_CHANGE`, five +`REQUIRES_EXECUTION`, and two `DROP` decisions. They still contained zero +truncation proposals. + +This is safer than v1, but it confirms that prompt wording alone did not give +GLM adequate recall for step-only salvage. + +## Truncation eligibility + +### Complete trajectory SFT + +Additional records repairable to `SFT_FULL` through truncation: **0/20**. + +A truncated failed or unverified trajectory lacks a verified terminal solution. +Treating it as full-SFT would convert absence of evidence into a success label. + +### Conservative step-only SFT + +High-confidence human-audited candidates: **4/20**. + +| Sample | Proposed cutoff region | Corrected next-step target | Why it is statically defensible | +|---:|---|---|---| +| 4 | After the observed no-match edge-case failure around turns 186-190 | Continue debugging the `undefined` result instead of discarding the failing case and claiming completion | The failure is present in the tool output before the corrected action | +| 6 | Before the unsupported completion claim at turn 181 | Inspect and implement the explicitly requested missing `MockRequest` scope, or state that the task remains incomplete | The missing scope appears in the user request, not only in the reference patch | +| 7 | Before the first prohibited `wrap_test.go` edit at turn 171 | Preserve the test file and inspect/refactor the production wrapper API so existing tests compile | The user explicitly prohibited test changes and the failing compiler output is already visible | +| 9 | Before the first prohibited `tests/integration/library.rs` edit at turn 95 | Keep tests unchanged and repair the production API/constructor path first | The no-test-edit constraint is explicit and visible before the bad action | + +These candidates may be used only as prefix-plus-corrected-next-action examples. +They must end at the corrected assistant/tool-call turn and contain no invented +tool result. + +Borderline candidates requiring human confirmation: samples 2, 3, 8, and 10. + +- Sample 2 created an issue reproduction but did not integrate it into the + rust-analyzer test harness before moving on. A useful next step exists, but + the standalone Rust file itself does not exercise the analyzer panic. +- Samples 3 and 8 have explicit failing test output and can teach failure + recovery, but several following turns are partially reasonable, so the exact + first irrecoverable assistant turn requires closer annotation. +- Sample 10 can be cut before the prohibited test edit at turn 85, but the + production patch is already semantically wrong for the target macOS branch, + making the retained context questionable for SFT. + +Under a conservative policy, use the confirmed count of four. Under a broader +error-recovery curriculum, the upper candidate count is eight, but the four +borderline cases should not be admitted automatically. + +## Model errors and reliability findings + +1. Sample 16 is `resolved=-1`, but GLM emitted the hard-fail code + `RESOLVED_ZERO_FOR_POSITIVE`. Sample 5 is `resolved=0`, but one pass omitted + that code. Local outcome mapping prevented an incorrect training upgrade, + but hard-fail code semantics need deterministic validation. +2. Classification quality was materially better than repair-generation + quality. The keep/reject boundary matched the manual review, while repair + plans contained type errors, decision/operation contradictions, and zero + truncation recall. +3. Long reasoning caused repeated gateway 502 responses. Turn-preserving + evidence compaction plus a low-latency retry completed all 20 + classifications. API transport failures must remain separate from QC + decisions. +4. A JSON-schema-valid plan is not sufficient. Operation-specific local policy + and deterministic dry-run are required before any mutation. + +## Recommended production decision + +- Admit the six accepted records to the silver `SFT_FULL` pool. +- Preserve the seven explicit negatives unchanged for DPO/error analysis. +- Drop samples 10 and 16 from positive/step training in their current form. +- Keep five unverified records in the execution-required pool. +- Create step-level candidates only from the four confirmed truncation cases, + then independently review their exact replacement messages and dry-run the + deterministic applier. +- Do not use GLM repair proposals without local policy enforcement and a second + reviewer. + +## Output files + +All files are under `outputs/glm52_sample20_20260805/`: + +- `classifications.jsonl`: 20 completed classifications +- `classification_errors_attempt1.jsonl`: archived gateway failures +- `repair_plans.jsonl`: 20 v1 repair plans +- `repair_plans_v2.jsonl`: 18 policy-valid v2 plans +- `repair_plan_v2_errors.jsonl`: two policy-rejected v2 proposals +- `repaired_static.jsonl`: one v1 dry-run mutation +- `apply_repair_errors.jsonl`: one v1 application failure diff --git a/scripts/merge_audits_in_source_order.py b/scripts/merge_audits_in_source_order.py new file mode 100644 index 0000000..57d04f9 --- /dev/null +++ b/scripts/merge_audits_in_source_order.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +"""Merge audit JSONL manifests into one unique source-ordered file.""" + +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path + + +def main() -> None: + """Merge the last result for each sample and write it in source order.""" + + parser = argparse.ArgumentParser() + parser.add_argument("source", type=Path, help="Source sample JSONL defining output order") + parser.add_argument("output", type=Path) + parser.add_argument("manifests", type=Path, nargs="+") + parser.add_argument("--require-count", type=int) + args = parser.parse_args() + + audits: dict[str, dict] = {} + for path in args.manifests: + if not path.exists(): + continue + with path.open(encoding="utf-8") as handle: + for line in handle: + if not line.strip(): + continue + value = json.loads(line) + sample_id = value.get("sample_id") + if not isinstance(sample_id, str) or not sample_id: + raise ValueError(f"Audit in {path} has no sample_id") + audits[sample_id] = value + + ordered: list[dict] = [] + with args.source.open(encoding="utf-8") as handle: + for line in handle: + if not line.strip(): + continue + source_record = json.loads(line) + sample_id = source_record.get("trajectory_id") or source_record.get("sample_id") + if sample_id in audits: + ordered.append(audits.pop(sample_id)) + + if audits: + raise ValueError(f"Found {len(audits)} audit IDs absent from the source sample") + if args.require_count is not None and len(ordered) != args.require_count: + raise ValueError(f"Expected {args.require_count} merged records, found {len(ordered)}") + + args.output.parent.mkdir(parents=True, exist_ok=True) + temporary = args.output.with_suffix(args.output.suffix + ".tmp") + with temporary.open("w", encoding="utf-8") as handle: + for value in ordered: + handle.write(json.dumps(value, ensure_ascii=False, default=str) + "\n") + os.replace(temporary, args.output) + print(json.dumps({"output": str(args.output), "records": len(ordered)}, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/sample_open_swe_filtered.py b/scripts/sample_open_swe_filtered.py new file mode 100644 index 0000000..872a1b0 --- /dev/null +++ b/scripts/sample_open_swe_filtered.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +"""Create a deterministic uniform sample from selected Open-SWE outcome classes.""" + +from __future__ import annotations + +import argparse +import json +import random +from pathlib import Path + +import pyarrow.parquet as pq + + +def main() -> None: + """Sample complete rows uniformly from records matching the requested outcomes.""" + + parser = argparse.ArgumentParser() + parser.add_argument("dataset_dir", type=Path) + parser.add_argument("output", type=Path) + parser.add_argument("--count", type=int, default=50) + parser.add_argument("--seed", type=int, default=20260806) + parser.add_argument( + "--resolved", + type=int, + nargs="+", + default=[0, -1], + help="Outcome values eligible for sampling; defaults to failed and unknown trajectories.", + ) + args = parser.parse_args() + + files = sorted(args.dataset_dir.glob("data/**/*.parquet")) + if not files: + raise SystemExit(f"No parquet files found under {args.dataset_dir}") + + allowed = set(args.resolved) + eligible: list[tuple[Path, int]] = [] + for path in files: + # Reading only the outcome column keeps the full-dataset eligibility pass inexpensive. + outcomes = pq.read_table(path, columns=["resolved"])["resolved"].to_pylist() + eligible.extend((path, index) for index, value in enumerate(outcomes) if value in allowed) + + if args.count > len(eligible): + raise SystemExit(f"Requested {args.count} rows from only {len(eligible)} eligible rows") + + rng = random.Random(args.seed) + chosen = rng.sample(eligible, args.count) + selected: list[dict] = [] + for path, file_row_index in chosen: + parquet_file = pq.ParquetFile(path) + row_group_start = 0 + for row_group_index in range(parquet_file.num_row_groups): + row_group_rows = parquet_file.metadata.row_group(row_group_index).num_rows + if file_row_index < row_group_start + row_group_rows: + table = parquet_file.read_row_group(row_group_index) + row = table.slice(file_row_index - row_group_start, 1).to_pylist()[0] + row["_sample"] = { + "sampling_method": "uniform without replacement over selected resolved values", + "resolved_values": sorted(allowed), + "seed": args.seed, + "source_file": str(path.relative_to(args.dataset_dir)), + "file_row_index": file_row_index, + "row_group_index": row_group_index, + } + selected.append(row) + break + row_group_start += row_group_rows + else: + raise RuntimeError(f"Could not locate row {file_row_index} in {path}") + + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.open("w", encoding="utf-8") as handle: + for row in selected: + handle.write(json.dumps(row, ensure_ascii=False, default=str) + "\n") + + outcome_counts = {str(value): 0 for value in sorted(allowed)} + for row in selected: + outcome_counts[str(row["resolved"])] += 1 + print( + json.dumps( + { + "output": str(args.output), + "seed": args.seed, + "eligible_rows": len(eligible), + "sample_rows": len(selected), + "sample_outcome_counts": outcome_counts, + }, + ensure_ascii=False, + indent=2, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/split_jsonl_remaining.py b/scripts/split_jsonl_remaining.py new file mode 100644 index 0000000..8b5f9d3 --- /dev/null +++ b/scripts/split_jsonl_remaining.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +"""Split unfinished JSONL records into balanced temporary retry shards.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + + +def record_id(value: dict) -> str: + """Return the stable trajectory identifier used by audit manifests.""" + + sample_id = value.get("trajectory_id") or value.get("sample_id") + if not isinstance(sample_id, str) or not sample_id: + raise ValueError("JSONL record has no trajectory_id/sample_id") + return sample_id + + +def main() -> None: + """Write unfinished input records round-robin across the requested shards.""" + + parser = argparse.ArgumentParser() + parser.add_argument("input", type=Path) + parser.add_argument("completed", type=Path) + parser.add_argument("output_dir", type=Path) + parser.add_argument("--shards", type=int, default=3) + args = parser.parse_args() + if args.shards < 1: + raise SystemExit("--shards must be positive") + + completed_ids: set[str] = set() + with args.completed.open(encoding="utf-8") as handle: + for line in handle: + if line.strip(): + completed_ids.add(record_id(json.loads(line))) + + args.output_dir.mkdir(parents=True, exist_ok=True) + handles = [ + (args.output_dir / f"part_{index}.input.jsonl").open("w", encoding="utf-8") + for index in range(args.shards) + ] + counts = [0] * args.shards + try: + unfinished = 0 + with args.input.open(encoding="utf-8") as handle: + for line in handle: + if not line.strip(): + continue + value = json.loads(line) + if record_id(value) in completed_ids: + continue + shard = unfinished % args.shards + handles[shard].write(line) + counts[shard] += 1 + unfinished += 1 + finally: + for handle in handles: + handle.close() + + print( + json.dumps( + { + "completed": len(completed_ids), + "unfinished": sum(counts), + "shard_counts": counts, + }, + indent=2, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/src/swe_data_processing/__init__.py b/src/swe_data_processing/__init__.py new file mode 100644 index 0000000..46275a5 --- /dev/null +++ b/src/swe_data_processing/__init__.py @@ -0,0 +1,6 @@ +"""Static quality-control utilities for the Open-SWE-Traces dataset.""" + +from .config import Settings + +__all__ = ["Settings"] +__version__ = "0.1.0" diff --git a/src/swe_data_processing/audit.py b/src/swe_data_processing/audit.py new file mode 100644 index 0000000..086ae1c --- /dev/null +++ b/src/swe_data_processing/audit.py @@ -0,0 +1,106 @@ +"""Local validation and scoring for GLM trajectory-quality audits.""" + +from __future__ import annotations + +from typing import Any + +from .features import index_trajectory +from .policy import PolicyViolation + +DIMENSION_WEIGHTS = { + "planning": 0.15, + "tool_selection": 0.20, + "observation_use": 0.20, + "efficiency": 0.15, + "verification_discipline": 0.20, + "claim_calibration": 0.10, +} + +SEVERITY_PENALTIES = {"MINOR": 1.0, "MAJOR": 5.0, "CRITICAL": 12.0} + + +def validate_audit(record: dict[str, Any], result: dict[str, Any]) -> None: + """Reject internally inconsistent or non-causal audit outputs.""" + + trajectory = index_trajectory(record) + assistant_turns = { + message["turn_id"] for message in trajectory if message.get("role") == "assistant" + } + all_turns = {message["turn_id"] for message in trajectory} + + first_bad_turn = result["truncation"]["first_bad_assistant_turn"] + truncate_before = result["truncation"]["truncate_before_turn"] + prefix_usable = result["truncation"]["prefix_usable"] + evaluation_mode = result["evaluation_mode"] + recommended_use = result["recommended_use"] + if evaluation_mode == "PROCESS_SALVAGE" and recommended_use == "FULL_TRAJECTORY_CANDIDATE": + raise PolicyViolation("A failed/unknown trajectory cannot be a full-trajectory candidate") + if evaluation_mode == "SUCCESS_QUALITY" and recommended_use == "PROCESS_PREFIX_CANDIDATE": + raise PolicyViolation("A successful trajectory is not evaluated as process salvage") + if prefix_usable != (recommended_use == "PROCESS_PREFIX_CANDIDATE"): + raise PolicyViolation("prefix_usable and recommended_use are inconsistent") + if first_bad_turn is not None and first_bad_turn not in assistant_turns: + raise PolicyViolation("first_bad_assistant_turn must reference an assistant turn") + if truncate_before is not None and truncate_before not in assistant_turns: + raise PolicyViolation("truncate_before_turn must reference an assistant turn") + if ( + evaluation_mode == "PROCESS_SALVAGE" + and prefix_usable + and (first_bad_turn is None or truncate_before is None) + ): + raise PolicyViolation("A usable process prefix requires an explicit bad assistant turn") + if first_bad_turn is not None and truncate_before != first_bad_turn: + raise PolicyViolation("Truncation must begin at the first bad assistant turn") + + start = result["truncation"]["acceptable_start_turn"] + end = result["truncation"]["acceptable_end_turn"] + if (start is None) != (end is None): + raise PolicyViolation("Acceptable turn range must be fully null or fully specified") + if start is not None and (start not in assistant_turns or end not in assistant_turns or start > end): + raise PolicyViolation("Acceptable turn range must reference ordered assistant turns") + + for issue in result["behavior_issues"]: + if issue["assistant_turn"] not in assistant_turns: + raise PolicyViolation("Every behavior issue must reference an assistant turn") + result_turn = issue["tool_result_turn"] + if result_turn is not None and result_turn not in all_turns: + raise PolicyViolation("tool_result_turn references a missing turn") + + counts = result["issue_counts"] + issues = result["behavior_issues"] + expected = { + "errors": sum(issue["kind"] == "ERROR" for issue in issues), + "inefficiencies": sum(issue["kind"] == "INEFFICIENCY" for issue in issues), + "critical": sum(issue["severity"] == "CRITICAL" for issue in issues), + "major": sum(issue["severity"] == "MAJOR" for issue in issues), + "minor": sum(issue["severity"] == "MINOR" for issue in issues), + } + if counts != expected: + raise PolicyViolation(f"issue_counts do not match behavior_issues: expected {expected}") + + +def compute_quality_score(result: dict[str, Any]) -> dict[str, Any]: + """Compute a deterministic educational-quality score from audit outputs.""" + + dimensions = result["quality_dimensions"] + base_score = sum(dimensions[name] * 20.0 * weight for name, weight in DIMENSION_WEIGHTS.items()) + issue_penalty = sum(SEVERITY_PENALTIES[issue["severity"]] for issue in result["behavior_issues"]) + issue_penalty = min(35.0, issue_penalty) + score = max(0.0, min(100.0, base_score - issue_penalty)) + + counts = result["issue_counts"] + if counts["critical"]: + tier = "LOW" + elif score >= 85: + tier = "HIGH" + elif score >= 70: + tier = "MEDIUM" + else: + tier = "LOW" + return { + "base_dimension_score": round(base_score, 2), + "issue_penalty": round(issue_penalty, 2), + "educational_quality_score": round(score, 2), + "quality_tier": tier, + "formula_version": "weighted-dimensions-minus-issue-severity-v1", + } diff --git a/src/swe_data_processing/cli.py b/src/swe_data_processing/cli.py new file mode 100644 index 0000000..893b74d --- /dev/null +++ b/src/swe_data_processing/cli.py @@ -0,0 +1,303 @@ +"""Command-line interface for static Open-SWE-Traces quality control.""" + +from __future__ import annotations + +import argparse +import json +import sys +from collections.abc import Callable +from pathlib import Path +from typing import Any + +from .client import GLMClient +from .config import Settings +from .features import extract_static_signals +from .io import append_jsonl, get_sample_id, iter_records, load_completed_ids, load_jsonl_index +from .repair import apply_static_repair +from .workflow import audit_trajectory, classify_record, plan_static_repair, review_repaired_record + + +def _write_error(path: Path | None, sample_id: str, stage: str, error: Exception) -> None: + """Write a compact non-sensitive error record for later retry.""" + + value = { + "sample_id": sample_id, + "stage": stage, + "error_type": type(error).__name__, + "error": str(error)[:2000], + } + if path is None: + print(json.dumps(value, ensure_ascii=False), file=sys.stderr) + else: + append_jsonl(path, value) + + +def _run_streaming_stage( + *, + input_path: Path, + output_path: Path, + errors_path: Path | None, + limit: int | None, + resume: bool, + stage_name: str, + processor: Callable[[dict[str, Any]], dict[str, Any] | None], +) -> int: + """Run one append-only stage with resume support and progress reporting.""" + + completed = load_completed_ids(output_path) if resume else set() + processed = 0 + failures = 0 + for record in iter_records(input_path): + sample_id = get_sample_id(record) + if sample_id in completed: + continue + if limit is not None and processed >= limit: + break + try: + result = processor(record) + if result is not None: + append_jsonl(output_path, result) + except Exception as exc: # noqa: BLE001 - each sample must fail independently. + failures += 1 + _write_error(errors_path, sample_id, stage_name, exc) + processed += 1 + if processed % 10 == 0: + print(f"{stage_name}: processed={processed} failures={failures}", file=sys.stderr) + print(f"{stage_name}: completed={processed} failures={failures}", file=sys.stderr) + return 1 if failures else 0 + + +def command_features(args: argparse.Namespace) -> int: + """Extract deterministic features without requiring an API key.""" + + return _run_streaming_stage( + input_path=args.input, + output_path=args.output, + errors_path=args.errors, + limit=args.limit, + resume=args.resume, + stage_name="features", + processor=lambda record: { + "sample_id": get_sample_id(record), + "resolved": record.get("resolved"), + "static_signals": extract_static_signals(record), + }, + ) + + +def command_classify(args: argparse.Namespace) -> int: + """Classify records through GLM-5.2 and local policy enforcement.""" + + settings = Settings.from_env() + with GLMClient(settings) as client: + return _run_streaming_stage( + input_path=args.input, + output_path=args.output, + errors_path=args.errors, + limit=args.limit, + resume=args.resume, + stage_name="classification", + processor=lambda record: classify_record(record, client), + ) + + +def command_audit(args: argparse.Namespace) -> int: + """Score educational process quality and locate causal truncation points.""" + + settings = Settings.from_env() + with GLMClient(settings) as client: + return _run_streaming_stage( + input_path=args.input, + output_path=args.output, + errors_path=args.errors, + limit=args.limit, + resume=args.resume, + stage_name="trajectory_audit", + processor=lambda record: audit_trajectory(record, client), + ) + + +def command_repair_plan(args: argparse.Namespace) -> int: + """Ask GLM-5.2 for allowlisted static repair plans.""" + + classifications = load_jsonl_index(args.classifications) + settings = Settings.from_env() + with GLMClient(settings) as client: + def processor(record: dict[str, Any]) -> dict[str, Any] | None: + sample_id = get_sample_id(record) + classification = classifications[sample_id] + if classification["qc_decision"] != "STATIC_REPAIR" and not args.include_nonrepair: + return None + return plan_static_repair(record, classification, client) + + return _run_streaming_stage( + input_path=args.input, + output_path=args.output, + errors_path=args.errors, + limit=args.limit, + resume=args.resume, + stage_name="repair_plan", + processor=processor, + ) + + +def command_apply_repair(args: argparse.Namespace) -> int: + """Apply validated static repair plans deterministically without API calls.""" + + plans = load_jsonl_index(args.plans) + completed = load_completed_ids(args.output) if args.resume else set() + processed = 0 + failures = 0 + for record in iter_records(args.input): + sample_id = get_sample_id(record) + if sample_id in completed or sample_id not in plans: + continue + if args.limit is not None and processed >= args.limit: + break + try: + repaired, diff = apply_static_repair(record, plans[sample_id]) + repaired["sample_id"] = sample_id + repaired["static_repair_diff"] = diff + append_jsonl(args.output, repaired) + except Exception as exc: # noqa: BLE001 + failures += 1 + _write_error(args.errors, sample_id, "apply_repair", exc) + processed += 1 + print(f"apply_repair: completed={processed} failures={failures}", file=sys.stderr) + return 1 if failures else 0 + + +def command_review(args: argparse.Namespace) -> int: + """Run independent GLM review of applied static repairs.""" + + classifications = load_jsonl_index(args.classifications) + plans = load_jsonl_index(args.plans) + repaired_records = load_jsonl_index(args.repaired) + settings = Settings.from_env() + with GLMClient(settings) as client: + def processor(record: dict[str, Any]) -> dict[str, Any] | None: + sample_id = get_sample_id(record) + if sample_id not in repaired_records: + return None + repaired = repaired_records[sample_id] + return review_repaired_record( + original_record=record, + classification=classifications[sample_id], + repair_plan=plans[sample_id], + repaired_record=repaired, + structured_diff=repaired.get("static_repair_diff") or {}, + client=client, + ) + + return _run_streaming_stage( + input_path=args.input, + output_path=args.output, + errors_path=args.errors, + limit=args.limit, + resume=args.resume, + stage_name="repair_review", + processor=processor, + ) + + +def command_smoke_test(_: argparse.Namespace) -> int: + """Perform one minimal authenticated structured-output request.""" + + settings = Settings.from_env() + schema = { + "type": "object", + "additionalProperties": False, + "required": ["status", "model"], + "properties": { + "status": {"const": "ok"}, + "model": {"type": "string", "minLength": 1}, + }, + } + with GLMClient(settings) as client: + response = client.invoke_json( + system_prompt=( + "Return a JSON object with status='ok' and model set to the model name you were asked to use." + ), + payload={"requested_model": settings.model}, + schema=schema, + ) + print( + json.dumps( + { + "endpoint": settings.endpoint, + "model": settings.model, + "response": response.data, + "request_id": response.request_id, + "usage": response.usage, + "compatibility_fallback_used": response.compatibility_fallback_used, + }, + ensure_ascii=False, + indent=2, + ) + ) + return 0 + + +def _add_stream_arguments(parser: argparse.ArgumentParser) -> None: + """Add arguments shared by streaming pipeline stages.""" + + parser.add_argument("--input", type=Path, required=True, help="JSONL file or dataset directory") + parser.add_argument("--output", type=Path, required=True, help="Append-only output JSONL") + parser.add_argument("--errors", type=Path, help="Optional error JSONL") + parser.add_argument("--limit", type=int, help="Maximum number of new records") + parser.add_argument("--resume", action="store_true", help="Skip IDs already present in output") + + +def build_parser() -> argparse.ArgumentParser: + """Construct the complete command-line parser.""" + + parser = argparse.ArgumentParser(prog="swe-qc", description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + + features = subparsers.add_parser("features", help="Extract deterministic static QC features") + _add_stream_arguments(features) + features.set_defaults(func=command_features) + + classify = subparsers.add_parser("classify", help="Classify trajectories through GLM-5.2") + _add_stream_arguments(classify) + classify.set_defaults(func=command_classify) + + audit = subparsers.add_parser( + "audit", help="Score process quality and locate causal truncation points" + ) + _add_stream_arguments(audit) + audit.set_defaults(func=command_audit) + + repair_plan = subparsers.add_parser("repair-plan", help="Create allowlisted static repair plans") + _add_stream_arguments(repair_plan) + repair_plan.add_argument("--classifications", type=Path, required=True) + repair_plan.add_argument("--include-nonrepair", action="store_true") + repair_plan.set_defaults(func=command_repair_plan) + + apply_repair = subparsers.add_parser("apply-repair", help="Apply static plans deterministically") + _add_stream_arguments(apply_repair) + apply_repair.add_argument("--plans", type=Path, required=True) + apply_repair.set_defaults(func=command_apply_repair) + + review = subparsers.add_parser("review", help="Independently review repaired records") + _add_stream_arguments(review) + review.add_argument("--classifications", type=Path, required=True) + review.add_argument("--plans", type=Path, required=True) + review.add_argument("--repaired", type=Path, required=True) + review.set_defaults(func=command_review) + + smoke_test = subparsers.add_parser("smoke-test", help="Test API authentication and JSON output") + smoke_test.set_defaults(func=command_smoke_test) + return parser + + +def main(argv: list[str] | None = None) -> int: + """CLI entry point used by the ``swe-qc`` console script.""" + + parser = build_parser() + args = parser.parse_args(argv) + return int(args.func(args)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/swe_data_processing/client.py b/src/swe_data_processing/client.py new file mode 100644 index 0000000..8da8b1a --- /dev/null +++ b/src/swe_data_processing/client.py @@ -0,0 +1,203 @@ +"""HTTP client for GLM-5.2 structured JSON generation.""" + +from __future__ import annotations + +import json +import random +import re +import time +from dataclasses import dataclass +from typing import Any + +import httpx +from jsonschema import ValidationError, validate + +from .config import Settings + + +class GLMClientError(RuntimeError): + """Raised when a GLM request cannot produce a valid structured response.""" + + +class GLMAuthenticationError(GLMClientError): + """Raised for non-retryable authentication and authorization failures.""" + + +@dataclass(frozen=True) +class GLMResponse: + """A validated model response with non-sensitive request metadata.""" + + data: dict[str, Any] + request_id: str | None + usage: dict[str, Any] | None + compatibility_fallback_used: bool + + +def _strip_json_fence(text: str) -> str: + """Remove a single Markdown JSON fence without changing JSON contents.""" + + stripped = text.strip() + match = re.fullmatch(r"```(?:json)?\s*(.*?)\s*```", stripped, flags=re.DOTALL | re.IGNORECASE) + return match.group(1).strip() if match else stripped + + +def _extract_message_text(response_data: dict[str, Any]) -> str: + """Extract text from common OpenAI-compatible response layouts.""" + + try: + content = response_data["choices"][0]["message"]["content"] + except (KeyError, IndexError, TypeError) as exc: + # The fallback keys make the client tolerant of lightweight gateway + # adapters while keeping the primary path OpenAI-compatible. + for key in ("output_text", "text", "output"): + value = response_data.get(key) + if isinstance(value, str): + return value + raise GLMClientError("The API response did not contain assistant message content") from exc + + if isinstance(content, str): + return content + if isinstance(content, list): + text_parts = [] + for block in content: + if isinstance(block, dict) and isinstance(block.get("text"), str): + text_parts.append(block["text"]) + if text_parts: + return "".join(text_parts) + raise GLMClientError("The assistant message content was not textual") + + +class GLMClient: + """Call GLM-5.2 with retry, compatibility fallback, and schema validation.""" + + def __init__(self, settings: Settings, *, transport: httpx.BaseTransport | None = None) -> None: + self.settings = settings + self._client = httpx.Client( + timeout=httpx.Timeout(settings.timeout_seconds), + transport=transport, + headers={ + "Authorization": f"Bearer {settings.api_key}", + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": "swe-data-processing/0.1.0", + }, + ) + + def close(self) -> None: + """Close the underlying HTTP connection pool.""" + + self._client.close() + + def __enter__(self) -> GLMClient: + return self + + def __exit__(self, *_: object) -> None: + self.close() + + def invoke_json( + self, + *, + system_prompt: str, + payload: dict[str, Any], + schema: dict[str, Any], + ) -> GLMResponse: + """Request one JSON object and validate it against ``schema``. + + Invalid JSON and schema violations are retried because model output can + occasionally be malformed. Authentication errors are never retried. + """ + + schema_text = json.dumps(schema, ensure_ascii=False, separators=(",", ":")) + full_system_prompt = ( + f"{system_prompt.rstrip()}\n\n" + "Return exactly one JSON object matching this JSON Schema:\n" + f"{schema_text}" + ) + compatibility_fallback = False + last_error: Exception | None = None + + for attempt in range(self.settings.max_retries): + request_payload = self._build_request( + full_system_prompt, + payload, + include_extensions=not compatibility_fallback, + ) + try: + response = self._client.post(self.settings.endpoint, json=request_payload) + request_id = response.headers.get("x-request-id") + + if response.status_code in {401, 403}: + raise GLMAuthenticationError( + f"Authentication failed with HTTP {response.status_code}; check GLM_API_KEY" + ) + if response.status_code == 400 and not compatibility_fallback: + # Some OpenAI-compatible gateways reject vendor-specific + # thinking parameters. Retry once with the portable subset. + compatibility_fallback = True + last_error = GLMClientError("The gateway rejected optional GLM parameters") + continue + if response.status_code == 429 or response.status_code >= 500: + raise httpx.HTTPStatusError( + f"Retryable API status {response.status_code}", + request=response.request, + response=response, + ) + response.raise_for_status() + + response_data = response.json() + raw_text = _extract_message_text(response_data) + parsed = json.loads(_strip_json_fence(raw_text)) + if not isinstance(parsed, dict): + raise GLMClientError("The model returned JSON that was not an object") + validate(instance=parsed, schema=schema) + usage = response_data.get("usage") + return GLMResponse( + data=parsed, + request_id=request_id, + usage=usage if isinstance(usage, dict) else None, + compatibility_fallback_used=compatibility_fallback, + ) + except GLMAuthenticationError: + raise + except (httpx.HTTPError, json.JSONDecodeError, ValidationError, GLMClientError) as exc: + last_error = exc + if attempt + 1 < self.settings.max_retries: + # Bounded exponential backoff avoids synchronized retries + # without making a single failed sample block indefinitely. + delay = min(20.0, 2.0**attempt) + random.uniform(0.0, 0.5) + time.sleep(delay) + + raise GLMClientError( + f"GLM failed to return valid structured output after {self.settings.max_retries} attempts: " + f"{type(last_error).__name__}: {last_error}" + ) + + def _build_request( + self, + system_prompt: str, + payload: dict[str, Any], + *, + include_extensions: bool, + ) -> dict[str, Any]: + """Build a portable chat-completions request body.""" + + request: dict[str, Any] = { + "model": self.settings.model, + "messages": [ + {"role": "system", "content": system_prompt}, + { + "role": "user", + "content": json.dumps(payload, ensure_ascii=False, separators=(",", ":")), + }, + ], + "temperature": self.settings.temperature, + "max_tokens": self.settings.max_tokens, + "response_format": {"type": "json_object"}, + "stream": False, + } + if include_extensions: + request["thinking"] = { + "type": "enabled" if self.settings.thinking_enabled else "disabled" + } + request["reasoning_effort"] = self.settings.reasoning_effort + return request diff --git a/src/swe_data_processing/config.py b/src/swe_data_processing/config.py new file mode 100644 index 0000000..a27151f --- /dev/null +++ b/src/swe_data_processing/config.py @@ -0,0 +1,93 @@ +"""Runtime configuration for the GLM API and local processing pipeline.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass + + +def _get_bool(name: str, default: bool) -> bool: + """Read a strict boolean environment variable with a safe default.""" + + raw_value = os.getenv(name) + if raw_value is None: + return default + normalized = raw_value.strip().lower() + if normalized in {"1", "true", "yes", "on"}: + return True + if normalized in {"0", "false", "no", "off"}: + return False + raise ValueError(f"{name} must be a boolean value, got {raw_value!r}") + + +def _get_int(name: str, default: int, minimum: int = 1) -> int: + """Read and validate an integer environment variable.""" + + value = int(os.getenv(name, str(default))) + if value < minimum: + raise ValueError(f"{name} must be at least {minimum}, got {value}") + return value + + +def _get_float(name: str, default: float, minimum: float = 0.0) -> float: + """Read and validate a floating-point environment variable.""" + + value = float(os.getenv(name, str(default))) + if value < minimum: + raise ValueError(f"{name} must be at least {minimum}, got {value}") + return value + + +@dataclass(frozen=True) +class Settings: + """Immutable settings used by API clients and command-line workflows.""" + + api_key: str + api_base: str = "https://llm-api.cowin.run" + api_path: str = "/v1/chat/completions" + model: str = "glm-5.2" + timeout_seconds: float = 300.0 + max_retries: int = 3 + max_tokens: int = 8192 + temperature: float = 0.0 + reasoning_effort: str = "high" + thinking_enabled: bool = True + + @property + def endpoint(self) -> str: + """Return the normalized absolute chat-completions URL.""" + + base = self.api_base.rstrip("/") + path = self.api_path if self.api_path.startswith("/") else f"/{self.api_path}" + return f"{base}{path}" + + @classmethod + def from_env(cls, *, require_api_key: bool = True) -> Settings: + """Construct settings from environment variables. + + The API key is intentionally loaded only from ``GLM_API_KEY``. The + project never reads a committed configuration file containing a key. + """ + + api_key = os.getenv("GLM_API_KEY", "").strip() + if require_api_key and not api_key: + raise ValueError("GLM_API_KEY is required but was not set") + + reasoning_effort = os.getenv("GLM_REASONING_EFFORT", "high").strip().lower() + if reasoning_effort not in {"low", "medium", "high", "max"}: + raise ValueError( + "GLM_REASONING_EFFORT must be one of: low, medium, high, max" + ) + + return cls( + api_key=api_key, + api_base=os.getenv("GLM_API_BASE", "https://llm-api.cowin.run"), + api_path=os.getenv("GLM_API_PATH", "/v1/chat/completions"), + model=os.getenv("GLM_MODEL", "glm-5.2"), + timeout_seconds=_get_float("GLM_TIMEOUT_SECONDS", 300.0, 1.0), + max_retries=_get_int("GLM_MAX_RETRIES", 3, 1), + max_tokens=_get_int("GLM_MAX_TOKENS", 8192, 1), + temperature=_get_float("GLM_TEMPERATURE", 0.0, 0.0), + reasoning_effort=reasoning_effort, + thinking_enabled=_get_bool("GLM_THINKING_ENABLED", True), + ) diff --git a/src/swe_data_processing/evidence.py b/src/swe_data_processing/evidence.py new file mode 100644 index 0000000..b202064 --- /dev/null +++ b/src/swe_data_processing/evidence.py @@ -0,0 +1,153 @@ +"""Prompt-only evidence compaction for long coding-agent trajectories. + +The functions in this module never modify the source dataset. They create a +bounded API representation while preserving turn identifiers, message roles, +commands, high-value verification evidence, and cryptographic hashes of every +shortened value. +""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + + +def _as_text(value: Any) -> str: + """Convert arbitrary JSON-compatible content into deterministic text.""" + + if isinstance(value, str): + return value + if value is None: + return "" + return json.dumps(value, ensure_ascii=False, sort_keys=True, default=str) + + +def compact_text(value: Any, max_chars: int) -> str: + """Return a head/tail preview with an auditable hash when text is long.""" + + text = _as_text(value) + if len(text) <= max_chars: + return text + digest = hashlib.sha256(text.encode("utf-8")).hexdigest() + marker = f"\n...[COMPACTED original_chars={len(text)} sha256={digest}]...\n" + available = max(2, max_chars - len(marker)) + head_chars = available // 3 + tail_chars = available - head_chars + return f"{text[:head_chars]}{marker}{text[-tail_chars:]}" + + +def compact_patch_object(value: Any, max_patch_chars: int = 30_000) -> dict[str, Any]: + """Compact only long string fields in a patch metadata object.""" + + if not isinstance(value, dict): + return {} + result: dict[str, Any] = {} + for key, item in value.items(): + if isinstance(item, str): + limit = max_patch_chars if key == "patch" else 4_000 + result[key] = compact_text(item, limit) + else: + result[key] = item + return result + + +def _compact_tool_call(call: Any, argument_limit: int) -> dict[str, Any]: + """Preserve tool identity and a bounded, hashed representation of arguments.""" + + if not isinstance(call, dict): + return {"malformed_call_preview": compact_text(call, argument_limit)} + function = call.get("function") + if not isinstance(function, dict): + return { + "id": call.get("id"), + "type": call.get("type"), + "malformed_function_preview": compact_text(function, argument_limit), + } + return { + "id": call.get("id"), + "type": call.get("type"), + "function": { + "name": function.get("name"), + "arguments": compact_text(function.get("arguments", "{}"), argument_limit), + }, + } + + +def compact_trajectory( + trajectory: list[dict[str, Any]], + static_signals: dict[str, Any], +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + """Build a bounded trajectory view that prioritizes decisive evidence. + + Test commands and observations, state-changing turns, malformed tool calls, + and the final eight turns receive larger previews. Other turns remain in the + prompt with their IDs and roles but use small previews, so ordering and + structural problems are still visible to the classifier. + """ + + important_turns = set(static_signals.get("stateful_turns") or []) + for event in static_signals.get("test_events") or []: + for key in ("command_turn", "output_turn"): + turn_id = event.get(key) + if isinstance(turn_id, int): + important_turns.add(turn_id) + for key in ("malformed_tool_turns", "unknown_tool_turns"): + for item in static_signals.get(key) or []: + turn_id = item.get("turn_id") if isinstance(item, dict) else None + if isinstance(turn_id, int): + important_turns.add(turn_id) + first_final_turn = max(1, len(trajectory) - 7) + important_turns.update(range(first_final_turn, len(trajectory) + 1)) + + compacted: list[dict[str, Any]] = [] + compacted_turns: list[int] = [] + for message in trajectory: + turn_id = int(message["turn_id"]) + role = message.get("role") + important = turn_id in important_turns + if role == "tool": + content_limit = 6_000 if important else 600 + elif role == "assistant": + content_limit = 4_000 if important else 800 + else: + content_limit = 2_000 if important else 800 + argument_limit = 6_000 if important else 1_500 + + original_content = _as_text(message.get("content")) + content = compact_text(original_content, content_limit) + if content != original_content: + compacted_turns.append(turn_id) + item: dict[str, Any] = { + "turn_id": turn_id, + "role": role, + "content": content, + } + if role == "assistant" and message.get("reasoning_content") is not None: + reasoning_limit = 3_000 if important else 700 + original_reasoning = _as_text(message.get("reasoning_content")) + item["reasoning_content"] = compact_text(original_reasoning, reasoning_limit) + if item["reasoning_content"] != original_reasoning: + compacted_turns.append(turn_id) + for key in ("name", "tool_call_id"): + if key in message: + item[key] = message.get(key) + if message.get("tool_calls") is not None: + item["tool_calls"] = [ + _compact_tool_call(call, argument_limit) + for call in (message.get("tool_calls") or []) + ] + compacted.append(item) + + metadata = { + "method": "turn-preserving-head-tail-v1", + "source_turn_count": len(trajectory), + "important_turn_ids": sorted(important_turns), + "content_compacted_turn_ids": compacted_turns, + "note": ( + "Compaction affects only the API prompt. Original dataset messages, " + "tool outputs, labels, and patches remain unchanged. Every shortened " + "value contains its original character count and SHA-256 hash." + ), + } + return compacted, metadata diff --git a/src/swe_data_processing/features.py b/src/swe_data_processing/features.py new file mode 100644 index 0000000..c714273 --- /dev/null +++ b/src/swe_data_processing/features.py @@ -0,0 +1,283 @@ +"""Deterministic feature extraction from agent trajectories and patches.""" + +from __future__ import annotations + +import json +import re +from collections import Counter +from typing import Any + +TEST_COMMAND_RE = re.compile( + r"(?:^|[;&\s])(pytest|py\.test|cargo\s+(?:test|check)|go\s+(?:test|build)|" + r"npm\s+(?:run\s+)?test|yarn\s+test|pnpm\s+test|mvn\s+(?:test|verify)|" + r"gradle\w*\s+test|make\s+(?:test|check)|ctest|phpunit|rspec)(?:\s|$)", + re.IGNORECASE, +) +FAILURE_RE = re.compile( + r"traceback|compilation failed|build fail(?:ed|ure)|test result:\s*failed|" + r"(?:^|\s)[1-9][0-9]*\s+(?:failed|failures?)\b|^\s*fail(?:\s|$)|" + r"panic:|timed? out|timeout|" + r"command not found|permission denied|exit (?:code|status):?\s*[1-9]", + re.IGNORECASE | re.MULTILINE, +) +SUCCESS_RE = re.compile( + r"test result:\s*ok|\bbuild success\b|\btests? passed\b|" + r"\bpassed\b|exit (?:code|status):?\s*0", + re.IGNORECASE, +) +EXIT_CODE_RE = re.compile(r"exit (?:code|status):?\s*(-?\d+)", re.IGNORECASE) +STATEFUL_SHELL_RE = re.compile( + r"(?:^|[;&|]\s*)(?:sed\s+-i|perl\s+-pi|tee\s+|cp\s+|mv\s+|rm\s+|" + r"git\s+(?:apply|checkout|restore|reset|add)|patch\s+|mkdir\s+|touch\s+)|" + r"(?:^|\s)(?:>|>>)(?:\s|\S)", + re.IGNORECASE, +) +TEST_CONSTRAINT_RE = re.compile( + r"(?:do not|don't|must not|should not)\s+(?:modify|change|edit).*?test", + re.IGNORECASE | re.DOTALL, +) +VENDOR_CONSTRAINT_RE = re.compile( + r"(?:only|minimal changes? to)\s+(?:non-test|source|production)\s+files?", + re.IGNORECASE, +) + + +def _message_content(message: dict[str, Any]) -> str: + """Convert a message content field to stable text for pattern matching.""" + + content = message.get("content") + if isinstance(content, str): + return content + if content is None: + return "" + return json.dumps(content, ensure_ascii=False, default=str) + + +def _parse_tool_definitions(raw_tools: list[Any]) -> tuple[set[str], list[int]]: + """Return allowed tool names and indexes of malformed definitions.""" + + names: set[str] = set() + malformed: list[int] = [] + for index, raw_tool in enumerate(raw_tools): + try: + tool = json.loads(raw_tool) if isinstance(raw_tool, str) else raw_tool + if not isinstance(tool, dict): + raise TypeError("tool definition is not an object") + function = tool.get("function") or tool + name = function.get("name") if isinstance(function, dict) else None + if isinstance(name, str) and name: + names.add(name) + else: + raise ValueError("tool definition has no name") + except (json.JSONDecodeError, TypeError, ValueError): + malformed.append(index) + return names, malformed + + +def _tool_call_details(call: dict[str, Any]) -> tuple[str, dict[str, Any] | None, str | None]: + """Parse a tool call into name, arguments, and an optional parse error.""" + + function = call.get("function") or {} + if not isinstance(function, dict): + return "", None, "function is not an object" + name = function.get("name") + raw_arguments = function.get("arguments", "{}") + if not isinstance(name, str): + name = "" + try: + arguments = json.loads(raw_arguments) if isinstance(raw_arguments, str) else raw_arguments + if not isinstance(arguments, dict): + return name, None, "arguments do not decode to an object" + return name, arguments, None + except json.JSONDecodeError as exc: + return name, None, f"{type(exc).__name__}: {exc}" + + +def _command_from_arguments(arguments: dict[str, Any] | None) -> str: + """Extract the command-like value used by common dataset tools.""" + + if not arguments: + return "" + for key in ("command", "cmd"): + value = arguments.get(key) + if isinstance(value, str): + return value + return "" + + +def _classify_test_level(command: str) -> str: + """Estimate test scope conservatively from a shell command.""" + + lowered = command.lower() + if "cargo check" in lowered or "go build" in lowered: + return "BUILD" + full_suite_tokens = ( + "go test ./...", + "npm test", + "cargo test", + "pytest tests", + "mvn test", + ) + if any(token in lowered for token in full_suite_tokens): + return "FULL" + if re.search(r"pytest\s+[^|;&]+::|go test\s+\S+/\.\.\.|cargo test\s+[-\w]*\s+\S+", lowered): + return "TARGETED" + return "MODULE" + + +def _extract_exit_codes(output: str) -> list[int]: + """Extract all explicit exit codes from a tool observation.""" + + return [int(value) for value in EXIT_CODE_RE.findall(output)] + + +def extract_patch_files(patch: str | None) -> list[str]: + """Extract normalized destination file paths from a unified diff.""" + + files: list[str] = [] + for match in re.finditer(r"^\+\+\+\s+(?:b/)?(.+)$", patch or "", flags=re.MULTILINE): + path = match.group(1).strip() + if path != "/dev/null" and path not in files: + files.append(path) + return files + + +def index_trajectory(record: dict[str, Any]) -> list[dict[str, Any]]: + """Return a trajectory copy with deterministic one-based turn identifiers.""" + + indexed = [] + for turn_id, message in enumerate(record.get("trajectory") or [], 1): + if isinstance(message, dict): + item = dict(message) + else: + item = {"role": "invalid", "content": str(message)} + item["turn_id"] = turn_id + indexed.append(item) + return indexed + + +def extract_instruction_constraints(trajectory: list[dict[str, Any]]) -> list[str]: + """Extract explicit file-modification constraints from user turns.""" + + user_text = "\n".join( + _message_content(message) + for message in trajectory + if message.get("role") == "user" + ) + constraints = [] + if TEST_CONSTRAINT_RE.search(user_text): + constraints.append("Do not modify test files.") + if VENDOR_CONSTRAINT_RE.search(user_text): + constraints.append("Limit changes to production/source files.") + return constraints + + +def extract_static_signals(record: dict[str, Any]) -> dict[str, Any]: + """Build deterministic QC signals without executing repository code.""" + + trajectory = index_trajectory(record) + allowed_tools, malformed_tool_definitions = _parse_tool_definitions(record.get("tools") or []) + malformed_calls: list[dict[str, Any]] = [] + unknown_calls: list[dict[str, Any]] = [] + tool_calls: list[dict[str, Any]] = [] + test_events: list[dict[str, Any]] = [] + stateful_turns: list[int] = [] + editor_paths: list[str] = [] + role_counts = Counter(message.get("role", "invalid") for message in trajectory) + + for index, message in enumerate(trajectory): + if message.get("role") != "assistant": + continue + for raw_call in message.get("tool_calls") or []: + if not isinstance(raw_call, dict): + malformed_calls.append({"turn_id": index + 1, "error": "tool call is not an object"}) + continue + name, arguments, parse_error = _tool_call_details(raw_call) + command = _command_from_arguments(arguments) + call_item = { + "turn_id": index + 1, + "tool_name": name, + "tool_call_id": raw_call.get("id"), + "command": command, + } + tool_calls.append(call_item) + if parse_error: + malformed_calls.append({"turn_id": index + 1, "tool_name": name, "error": parse_error}) + if allowed_tools and name not in allowed_tools: + unknown_calls.append({"turn_id": index + 1, "tool_name": name}) + + editor_command = arguments.get("command") if arguments and name == "str_replace_editor" else None + editor_path = arguments.get("path") if arguments else None + if name == "str_replace_editor" and editor_command in {"create", "str_replace", "insert"}: + stateful_turns.append(index + 1) + if isinstance(editor_path, str) and editor_path not in editor_paths: + editor_paths.append(editor_path) + if command and STATEFUL_SHELL_RE.search(command): + stateful_turns.append(index + 1) + + if command and TEST_COMMAND_RE.search(command): + output = "" + output_turn = None + if index + 1 < len(trajectory) and trajectory[index + 1].get("role") == "tool": + output = _message_content(trajectory[index + 1]) + output_turn = index + 2 + explicit_codes = _extract_exit_codes(output) + masked_pipeline = "|" in command and "pipefail" not in command + explicit_failure = bool(FAILURE_RE.search(output)) or any( + code != 0 for code in explicit_codes + ) + explicit_success = bool(SUCCESS_RE.search(output)) and not explicit_failure + test_events.append( + { + "command_turn": index + 1, + "output_turn": output_turn, + "command": command, + "level": _classify_test_level(command), + "masked_pipeline": masked_pipeline, + "explicit_exit_codes": explicit_codes, + "explicit_failure": explicit_failure, + "explicit_success": explicit_success, + "output_preview": output[-1500:], + } + ) + + # The dataset alternates assistant calls and tool observations. The check + # remains conservative because some valid messages can contain no tool call. + alternation_issues = [] + for index in range(2, len(trajectory)): + expected = "assistant" if index % 2 == 0 else "tool" + actual = trajectory[index].get("role") + if actual != expected: + alternation_issues.append({"turn_id": index + 1, "expected": expected, "actual": actual}) + + metadata = record.get("metadata") or {} + model_patch = metadata.get("model_patch") or {} + reference_patch = metadata.get("reference_patch") or {} + model_patch_text = model_patch.get("patch") or "" + reference_patch_text = reference_patch.get("patch") or "" + model_files = extract_patch_files(model_patch_text) + reference_files = extract_patch_files(reference_patch_text) + + return { + "allowed_tool_names": sorted(allowed_tools), + "malformed_tool_definition_indexes": malformed_tool_definitions, + "malformed_tool_turns": malformed_calls, + "unknown_tool_turns": unknown_calls, + "alternation_issues": alternation_issues, + "role_counts": dict(role_counts), + "tool_call_count": len(tool_calls), + "stateful_turns": sorted(set(stateful_turns)), + "last_stateful_turn": max(stateful_turns, default=None), + "editor_paths": editor_paths, + "test_events": test_events, + "model_patch_files": model_files, + "reference_patch_files": reference_files, + "model_patch_chars": len(model_patch_text), + "reference_patch_chars": len(reference_patch_text), + "patch_size_ratio_to_reference": ( + round(len(model_patch_text) / len(reference_patch_text), 4) + if reference_patch_text + else None + ), + "instruction_constraints": extract_instruction_constraints(trajectory), + } diff --git a/src/swe_data_processing/io.py b/src/swe_data_processing/io.py new file mode 100644 index 0000000..5f8c04c --- /dev/null +++ b/src/swe_data_processing/io.py @@ -0,0 +1,94 @@ +"""Streaming readers and append-only JSONL writers for dataset workflows.""" + +from __future__ import annotations + +import json +from collections.abc import Iterator +from pathlib import Path +from typing import Any + +import pyarrow.parquet as pq + + +def get_sample_id(record: dict[str, Any]) -> str: + """Return the stable trajectory identifier used by pipeline manifests.""" + + value = record.get("trajectory_id") or record.get("sample_id") + if not isinstance(value, str) or not value: + raise ValueError("Record is missing a non-empty trajectory_id/sample_id") + return value + + +def iter_jsonl(path: Path) -> Iterator[dict[str, Any]]: + """Yield JSON objects from a UTF-8 JSONL file with line-aware errors.""" + + with path.open("r", encoding="utf-8") as handle: + for line_number, line in enumerate(handle, 1): + if not line.strip(): + continue + try: + value = json.loads(line) + except json.JSONDecodeError as exc: + raise ValueError(f"Invalid JSON at {path}:{line_number}: {exc}") from exc + if not isinstance(value, dict): + raise ValueError(f"Expected a JSON object at {path}:{line_number}") + yield value + + +def iter_parquet_dataset(dataset_dir: Path, *, batch_size: int = 16) -> Iterator[dict[str, Any]]: + """Stream records from every Open-SWE-Traces Parquet shard in stable order.""" + + files = sorted(dataset_dir.glob("data/**/*.parquet")) + if not files: + raise FileNotFoundError(f"No Parquet shards found under {dataset_dir / 'data'}") + for path in files: + parquet_file = pq.ParquetFile(path) + for batch in parquet_file.iter_batches(batch_size=batch_size): + for record in batch.to_pylist(): + if isinstance(record, dict): + yield record + + +def iter_records(path: Path) -> Iterator[dict[str, Any]]: + """Read either a JSONL file or a local Hugging Face dataset directory.""" + + if path.is_dir(): + yield from iter_parquet_dataset(path) + elif path.suffix.lower() in {".jsonl", ".json"}: + yield from iter_jsonl(path) + else: + raise ValueError(f"Unsupported input path: {path}") + + +def append_jsonl(path: Path, value: dict[str, Any]) -> None: + """Append one compact JSON object and flush it for crash-safe progress.""" + + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(value, ensure_ascii=False, default=str) + "\n") + handle.flush() + + +def load_completed_ids(path: Path) -> set[str]: + """Load sample IDs already present in an append-only output manifest.""" + + if not path.exists(): + return set() + completed = set() + for value in iter_jsonl(path): + sample_id = value.get("sample_id") + if isinstance(sample_id, str) and sample_id: + completed.add(sample_id) + return completed + + +def load_jsonl_index(path: Path) -> dict[str, dict[str, Any]]: + """Load a moderate JSONL manifest into memory keyed by sample ID.""" + + index = {} + for value in iter_jsonl(path): + sample_id = get_sample_id(value) + if sample_id in index: + raise ValueError(f"Duplicate sample ID {sample_id!r} in {path}") + index[sample_id] = value + return index diff --git a/src/swe_data_processing/policy.py b/src/swe_data_processing/policy.py new file mode 100644 index 0000000..82f3d0f --- /dev/null +++ b/src/swe_data_processing/policy.py @@ -0,0 +1,89 @@ +"""Deterministic policy guards applied after every GLM classification.""" + +from __future__ import annotations + +from typing import Any + + +class PolicyViolation(ValueError): + """Raised when model output violates a non-negotiable QC invariant.""" + + +def enforce_classification_policy(record: dict[str, Any], result: dict[str, Any]) -> None: + """Reject classifications that illegally upgrade evidence or outcomes.""" + + resolved = int(record.get("resolved", -1)) + decision = result["qc_decision"] + training_use = result["training_use"] + qc_passed = result["qc_passed"] + + expected_source_class = { + 1: "POSITIVE_CANDIDATE", + 0: "EXPLICIT_NEGATIVE", + -1: "UNVERIFIED", + }.get(resolved) + if expected_source_class is None: + raise PolicyViolation(f"Unsupported resolved value: {resolved}") + if result["source_outcome_class"] != expected_source_class: + raise PolicyViolation("source_outcome_class does not match the immutable resolved label") + + if resolved == 0 and (decision == "ACCEPT_SILVER_POSITIVE" or training_use == "SFT_FULL"): + raise PolicyViolation("resolved=0 cannot be promoted to a successful full SFT sample") + if resolved == -1 and training_use == "SFT_FULL": + raise PolicyViolation("resolved=-1 cannot be promoted to SFT_FULL without execution") + if qc_passed != (decision == "ACCEPT_SILVER_POSITIVE"): + raise PolicyViolation("qc_passed must be true only for ACCEPT_SILVER_POSITIVE") + + if decision == "ACCEPT_SILVER_POSITIVE": + if resolved != 1: + raise PolicyViolation("Only resolved=1 can be accepted as a silver positive") + if result["hard_fail_codes"]: + raise PolicyViolation("A silver positive cannot contain a hard-fail code") + if result["verification"]["status"] != "PASS_RELIABLE": + raise PolicyViolation("A silver positive requires reliable passing verification") + if any(value != "PASS" for value in result["dimensions"].values()): + raise PolicyViolation("Every QC dimension must pass for a silver positive") + + +def enforce_repair_policy(result: dict[str, Any]) -> None: + """Ensure a repair plan explicitly preserves all immutable evidence.""" + + invariants = result["invariants"] + if not all(invariants.values()): + raise PolicyViolation("Every static-repair invariant must be true") + if not result["requires_second_review"]: + raise PolicyViolation("Every repair plan requires independent second review") + + creates_step_example = result["repair_decision"] == "CREATE_STEP_EXAMPLE" + if creates_step_example and result["maximum_training_use"] != "SFT_STEP_ONLY": + raise PolicyViolation("A truncated or corrected step example cannot exceed SFT_STEP_ONLY") + + decision = result["repair_decision"] + operations = result["operations"] + if decision in {"NO_CHANGE", "REQUIRES_EXECUTION", "DROP"} and operations: + raise PolicyViolation(f"{decision} must not contain mutation operations") + if decision in {"APPLY_STATIC_REPAIR", "CREATE_STEP_EXAMPLE"} and not operations: + raise PolicyViolation(f"{decision} requires at least one operation") + if result["maximum_training_use"] == "DPO_REJECTED" and decision == "APPLY_STATIC_REPAIR": + raise PolicyViolation("Authentic DPO rejected trajectories must not be rewritten") + + operation_names = [operation["op"] for operation in operations] + if creates_step_example: + if "EMIT_STRUCTURALLY_CORRECT_ASSISTANT_TURN" not in operation_names: + raise PolicyViolation("A step example requires a corrected assistant turn") + if "TRUNCATE_BEFORE_TURN" not in operation_names: + raise PolicyViolation("A step example requires truncation after the corrected turn") + + for operation in operations: + op = operation["op"] + replacement = operation["replacement"] + if op in {"REMOVE_DUPLICATE_PROSE", "REWRITE_FINAL_SUMMARY", "REDACT_SECRET"}: + if not isinstance(replacement, str): + raise PolicyViolation(f"{op} requires a string replacement") + if op in { + "CANONICALIZE_JSON", + "NORMALIZE_TOOL_ALIAS", + "EMIT_STRUCTURALLY_CORRECT_ASSISTANT_TURN", + }: + if not isinstance(replacement, dict) or replacement.get("role") != "assistant": + raise PolicyViolation(f"{op} requires a complete assistant-message object") diff --git a/src/swe_data_processing/prompts/classification.md b/src/swe_data_processing/prompts/classification.md new file mode 100644 index 0000000..c760a81 --- /dev/null +++ b/src/swe_data_processing/prompts/classification.md @@ -0,0 +1,37 @@ +You are a conservative static quality auditor for Open-SWE-Traces. You cannot execute code, and you must not +assume any test result that is absent from the supplied evidence. Your task is classification only; do not +repair the sample. + +Mandatory policy: + +1. A sample with `resolved=0` can never become a successful full-trajectory SFT example. +2. A sample with `resolved=-1` cannot be promoted to a successful positive solely from your code intuition. +3. Only `resolved=1` with every static QC requirement satisfied may be `ACCEPT_SILVER_POSITIVE`. +4. Explicit test failures, compilation failures, timeouts, or trustworthy non-zero exit codes override an + assistant claim that the task succeeded. +5. Verification must occur after the final code-changing action. A later code change invalidates earlier tests. +6. A pipeline such as `test | grep`, `test | head`, `test | tail`, or `test | tee` has an untrustworthy exit + code unless `pipefail` or the left-side exit code is explicitly captured. +7. Build, syntax, and type checks are not functional tests. +8. A model patch need not be identical to the reference patch. Judge whether it addresses the same requirement + and core logic; do not reject solely because file sets differ. +9. Every critical conclusion must cite a turn ID or an exact metadata/patch field and include a quote no longer + than 240 characters. +10. If evidence is missing or conflicting, use `HOLD_UNVERIFIED`; do not guess. +11. `qc_passed` may be true only for `ACCEPT_SILVER_POSITIVE`. +12. The supplied trajectory is a turn-preserving API evidence view. A + `[COMPACTED ...]` marker is not dataset corruption and must not itself cause + a trajectory-integrity failure. Use the retained head/tail text, turn IDs, + static signals, character counts, and hashes; choose `HOLD_UNVERIFIED` when + decisive evidence falls outside a preview. + +Decision order: + +A. Map the immutable `resolved` value to `source_outcome_class`. +B. Validate trajectory and tool structure. +C. Locate the final state-changing turn. +D. Evaluate the final relevant verification after that turn, including scope and exit-code trustworthiness. +E. Compare files observed in tool calls, git status, assistant claims, and the model patch. +F. Check user constraints, patch contamination, and final-claim alignment. +G. Use the reference patch only as non-binding semantic evidence. +H. Return one schema-valid JSON object with decision, training use, failures, warnings, and cited evidence. diff --git a/src/swe_data_processing/prompts/repair.md b/src/swe_data_processing/prompts/repair.md new file mode 100644 index 0000000..21b673f --- /dev/null +++ b/src/swe_data_processing/prompts/repair.md @@ -0,0 +1,47 @@ +You are a static repair planner for Open-SWE-Traces. You have no execution environment. You may propose only +allowlisted repairs that preserve every execution fact, and you must return exactly one schema-valid JSON object. + +Immutable facts: + +- Do not change `resolved`, the model patch, or the reference patch. +- Preserve every existing tool output byte-for-byte. +- Do not create shell, test, build, lint, or tool results. +- Do not change tool-call semantics while retaining the old tool result. +- `resolved=0` cannot be repaired into a successful positive example. +- `resolved=-1` cannot be upgraded to `SFT_FULL`. +- Truncation or a corrected next action has a maximum use of `SFT_STEP_ONLY`. +- A rejected trajectory must remain unchanged when retained as an authentic + `DPO_REJECTED` example. Do not rewrite its failure or overconfidence away. + +The only allowed operations are: `CANONICALIZE_JSON`, `NORMALIZE_TOOL_ALIAS`, `REDACT_SECRET`, +`REMOVE_DUPLICATE_PROSE`, `REWRITE_FINAL_SUMMARY`, `DROP_REDUNDANT_READ_ONLY_PAIR`, +`TRUNCATE_BEFORE_TURN`, and `EMIT_STRUCTURALLY_CORRECT_ASSISTANT_TURN`. + +For every proposed operation, provide exact target turns, machine-checkable preconditions, an exact replacement, +and a reason. If a precondition cannot be proven from the input, do not propose the operation. + +Choose `REQUIRES_EXECUTION` or `DROP` instead of full-trajectory static repair when the problem involves code +semantics, failed tests, timeouts, missing dependencies, patch/trajectory inconsistency, prohibited test-file +changes, removal of stateful operations, source-code changes, or any need to create a new tool result. + +Step-only truncation is narrower than full-trajectory repair. Use `CREATE_STEP_EXAMPLE` only when all of the +following are true: + +1. A specific bad assistant turn is identifiable from evidence already visible before that turn, such as an + ignored test failure, deletion of an unexecuted reproduction, a prohibited test-file edit, or an unsupported + success claim. +2. The correct immediate next assistant action is uniquely defensible from that prior evidence without using the + hidden outcome or reference patch as privileged information. +3. The replacement is a complete assistant message and contains at most a tool call or an honest diagnostic + response. It must not claim the task is solved and must not include a fabricated tool result. +4. Operations first replace the bad assistant turn with `EMIT_STRUCTURALLY_CORRECT_ASSISTANT_TURN`, then use + `TRUNCATE_BEFORE_TURN` on the following turn. The resulting record ends at the corrected assistant action. +5. `maximum_training_use` is exactly `SFT_STEP_ONLY`. + +Do not truncate merely to hide a wrong patch, failed test, or external `resolved=0` label. If no uniquely correct +next action exists, retain the authentic negative, require execution, or drop the record. + +Decisions `NO_CHANGE`, `REQUIRES_EXECUTION`, and `DROP` must have an empty operations array. `APPLY_STATIC_REPAIR` +and `CREATE_STEP_EXAMPLE` must have at least one operation. + +Before returning, confirm all evidence-preservation invariants are true and require independent second review. diff --git a/src/swe_data_processing/prompts/review.md b/src/swe_data_processing/prompts/review.md new file mode 100644 index 0000000..98282db --- /dev/null +++ b/src/swe_data_processing/prompts/review.md @@ -0,0 +1,16 @@ +You are an independent reviewer of a static Open-SWE-Traces repair. You cannot execute code. Compare the original +sample, classification, repair plan, repaired sample, and structured diff. Approval means only that the static +repair is faithful, safe, and structurally valid; it does not prove code correctness. + +Reject the repair if any of the following is true: + +- `resolved`, the model patch, or the reference patch changed. +- Any tool output changed or a synthetic execution result was added. +- A file write, git operation, installation, service start, or other stateful turn was removed. +- Removal created a dangling tool pair or invalidated later state dependencies. +- Tool-call semantics changed without a newly executed result. +- A `resolved=0` or `resolved=-1` sample was upgraded to `SFT_FULL`. +- A truncated sample was represented as a complete successful trajectory. +- Training use exceeds the maximum supported by the original evidence and classification. + +Every invariant violation must set `approved=false`. Return exactly one JSON object matching the review schema. diff --git a/src/swe_data_processing/prompts/trajectory_audit.md b/src/swe_data_processing/prompts/trajectory_audit.md new file mode 100644 index 0000000..c8ae76f --- /dev/null +++ b/src/swe_data_processing/prompts/trajectory_audit.md @@ -0,0 +1,121 @@ +You are auditing the educational process quality of an Open-SWE-Traces coding-agent trajectory. You cannot run +code and must not repair or rewrite the trajectory. Return exactly one schema-valid JSON object. + +There are two related tasks: + +1. Identify clear erroneous or inefficient assistant/tool-call behavior using only evidence visible in the + trajectory at that point. +2. For a trajectory that is not externally marked successful, decide whether a prefix before the first clearly + bad assistant turn still teaches useful problem analysis, tool use, or observation interpretation. + +## Exact truncation-boundary definition + +For `PROCESS_SALVAGE`, scan assistant turns in chronological order and choose a boundary `B` only if all of the +following are true: + +1. The retained prefix is every message with `turn_id < B`; turn `B` and every later message are discarded. +2. Turn `B` is an assistant turn, and the action or reasoning at `B` is the earliest *major or critical, + unrecovered* defect that should not be taught as positive process data. +3. The prefix before `B` is still coherent and educational: it contains useful investigation, tool use, or correct + interpretation of observations, with at most a small number of minor or recovered defects. +4. Keeping turn `B` would materially lower the educational validity of the prefix. If keeping `B` is still + reasonable exploration, move the boundary later or return null. + +Use this decision procedure for tool interactions: + +- If an assistant chooses an invalid command, violates an explicit constraint, performs an unsafe state-changing + action, or targets a file known to be wrong, `B` is that assistant turn. +- If a reasonable tool call returns an error, nonzero exit status, timeout, or failed test, the tool result is not + a boundary. Keep it when it teaches useful diagnosis. If the next assistant ignores, contradicts, or falsely + explains that result, `B` is that next assistant turn. +- If the assistant makes a plausible hypothesis that is later disproved and then correctly adapts, do not truncate + at the hypothesis. Normal exploration and recovered mistakes are allowed. +- If an assistant starts a no-progress loop, `B` is the first assistant turn where repetition without meaningful + adaptation becomes clear, not the first failed attempt. +- If the implementation is useful but the final answer claims unsupported success, `B` is the assistant turn that + makes the unsupported claim; preserve the useful work before it. + +Before returning a non-null `B`, perform this two-sided check: + +- `B-1 check`: the last assistant action before `B` can still be shown to an SFT learner without teaching a major + known error. +- `B check`: quote visible evidence proving why assistant turn `B` itself must be excluded. + +Also perform a retained-prefix terminal-state check. The prefix immediately before `B` must not leave an earlier +major/critical episode open. Reject or move the boundary when the retained prefix contains an unreverted wrong-file +edit, constraint violation, destructive action, known patch pollution, repeated failing state-changing call, or +other harmful repository state whose recovery occurs at turn `B` or later. Never truncate immediately before a +rollback, cleanup, correction, or successful adaptation when doing so would preserve the mistake but discard its +recovery. Either move the boundary before the original harmful action or include the recovery and search later. + +An earlier-than-optimal boundary is acceptable when it is conservative: the retained prefix is coherent, has +meaningful educational signal, and contains no unrecovered major/critical behavior. Prefer safety over maximizing +length. Do not reject a safe candidate merely because useful later work would be omitted; that is yield loss, not +training-data corruption. + +If either side cannot be supported from the supplied trajectory, return a null boundary and `prefix_usable=false`. +Do not invent a boundary merely because the external outcome is unsuccessful. + +Causal rules: + +- Do not use the external outcome label or reference patch to claim an earlier action was wrong. The outcome label + only selects the evaluation mode. +- A failed tool call is not automatically bad. It may be a useful probe when its result is read correctly and the + next action makes progress. +- Mark an ERROR only for visible behavior such as ignoring a failure, contradicting an observation, violating an + explicit constraint, using an invalid argument, targeting the wrong file after contrary evidence, making an + unsafe state-changing action, or claiming unsupported success. +- Mark an INEFFICIENCY only when evidence is strong: essentially duplicate reads, repeated identical failures + without adaptation, long no-progress loops, unnecessary re-verification, or repetitive summaries. +- Do not punish normal exploration, one failed search, or a reasonable hypothesis that is corrected later. +- Distinguish the first observable minor defect from the first disqualifying turn. A valid process prefix may + contain a small number of recovered errors or inefficiencies. Truncate only before the first major/critical + assistant action that makes the remaining suffix unsuitable to teach. In other words, find the first turn after + which the process should no longer be shown as positive training data, not merely the first imperfect turn. +- The truncation turn must be an assistant turn. Tool output can be evidence, but truncation occurs before the + assistant action that mishandles prior evidence. +- A process prefix is usable only when the prefix before that assistant turn is coherent and contains meaningful + learning signal. It need not solve the full task and need not be perfectly efficient. +- If no causal first-bad turn can be proven, return null rather than using hidden outcome knowledge. +- A prompt `[COMPACTED ...]` marker is not trajectory corruption. If decisive text is unavailable, lower confidence + or return null. +- Every non-null truncation boundary and acceptable-range endpoint must be a `turn_id` whose role is `assistant`. +- `acceptable_start_turn` and `acceptable_end_turn` are a narrow uncertainty interval around `B`, not the retained + prefix range. Both endpoints must be actual assistant `turn_id` values near `B`. Use identical endpoints when the + boundary is clear. Never use the first trajectory turn as a default range start. +- In `SUCCESS_QUALITY` mode, audit the complete trajectory. Set every truncation boundary/range field to null and + set `prefix_usable` to false because process salvage is not applicable. +- In `PROCESS_SALVAGE` mode, behavior issues, issue counts, and quality dimensions must describe only the retained + prefix strictly before `truncate_before_turn`. The truncation evidence may separately describe the excluded bad + turn. This makes the score a score of the candidate training prefix, not of the discarded suffix. + +Quality dimensions are integers from 0 to 5: + +- planning: decomposition and hypothesis quality; +- tool_selection: appropriate tools, commands, and targets; +- observation_use: reads and responds to tool evidence correctly; +- efficiency: avoids redundant/no-progress work; +- verification_discipline: uses relevant checks and interprets their status honestly; +- claim_calibration: summaries match what was actually observed. + +Recommended-use rules: + +- `FULL_TRAJECTORY_CANDIDATE`: externally successful and no critical visible process defect; +- `PROCESS_PREFIX_CANDIDATE`: not successful, but a causal usable prefix and bad-turn boundary are identified; +- `HOLD`: evidence is insufficient or ambiguous; +- `REJECT`: no meaningful safe prefix or severe bad behavior starts too early. + +List only defensible behavior issues. Quotes must be short and copied from the supplied evidence. Do not propose +edits, corrected tool calls, or synthetic results. + +Final self-check before emitting JSON: + +- every boundary, range endpoint, and `behavior_issues[].assistant_turn` names an assistant turn; +- every `tool_result_turn` names an existing tool-result turn or is null; +- each evidence object contains exactly `turn_id` and `quote` (never use a `content` field), and every behavior issue + contains every schema-required field including `reason`; +- for process salvage, all listed behavior issues and dimension scores describe only retained messages before `B`; +- `first_bad_assistant_turn == truncate_before_turn == B` when non-null; +- a usable prefix has a non-null `B`; otherwise all four boundary/range values are null. +- the retained prefix has no open major/critical error episode or harmful state mutation awaiting recovery at or + after the boundary. diff --git a/src/swe_data_processing/repair.py b/src/swe_data_processing/repair.py new file mode 100644 index 0000000..455c070 --- /dev/null +++ b/src/swe_data_processing/repair.py @@ -0,0 +1,183 @@ +"""Deterministic application of allowlisted static repair operations.""" + +from __future__ import annotations + +import copy +import hashlib +import json +from typing import Any + +from .features import STATEFUL_SHELL_RE +from .io import get_sample_id +from .policy import PolicyViolation, enforce_repair_policy + + +def _hash(value: Any) -> str: + """Return a stable SHA-256 digest for audit records.""" + + encoded = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _assistant_command(message: dict[str, Any]) -> str: + """Extract a shell command from the first tool call in an assistant message.""" + + calls = message.get("tool_calls") or [] + if not calls or not isinstance(calls[0], dict): + return "" + function = calls[0].get("function") or {} + raw_arguments = function.get("arguments", "{}") if isinstance(function, dict) else "{}" + try: + arguments = json.loads(raw_arguments) if isinstance(raw_arguments, str) else raw_arguments + except json.JSONDecodeError: + return "" + return str(arguments.get("command") or arguments.get("cmd") or "") if isinstance(arguments, dict) else "" + + +def _is_read_only_pair(assistant_message: dict[str, Any], tool_message: dict[str, Any]) -> bool: + """Conservatively determine whether a tool pair has no repository side effects.""" + + if assistant_message.get("role") != "assistant" or tool_message.get("role") != "tool": + return False + calls = assistant_message.get("tool_calls") or [] + if len(calls) != 1 or not isinstance(calls[0], dict): + return False + function = calls[0].get("function") or {} + name = function.get("name") if isinstance(function, dict) else None + if name == "str_replace_editor": + raw_arguments = function.get("arguments", "{}") + try: + arguments = json.loads(raw_arguments) if isinstance(raw_arguments, str) else raw_arguments + except json.JSONDecodeError: + return False + return isinstance(arguments, dict) and arguments.get("command") == "view" + command = _assistant_command(assistant_message) + return bool(command) and not STATEFUL_SHELL_RE.search(command) + + +def _replace_message( + entry: dict[str, Any], + operation: str, + replacement: Any, +) -> None: + """Apply a non-deleting operation to one annotated trajectory entry.""" + + message = entry["message"] + if message.get("role") == "tool": + raise PolicyViolation(f"{operation} cannot modify a tool observation") + + if operation in {"REMOVE_DUPLICATE_PROSE", "REWRITE_FINAL_SUMMARY", "REDACT_SECRET"}: + if not isinstance(replacement, str): + raise PolicyViolation(f"{operation} requires a string replacement") + if message.get("role") != "assistant": + raise PolicyViolation(f"{operation} is restricted to assistant messages") + message["content"] = replacement + return + + if operation in { + "CANONICALIZE_JSON", + "NORMALIZE_TOOL_ALIAS", + "EMIT_STRUCTURALLY_CORRECT_ASSISTANT_TURN", + }: + if not isinstance(replacement, dict) or replacement.get("role") != "assistant": + raise PolicyViolation(f"{operation} requires a complete assistant-message object") + entry["message"] = copy.deepcopy(replacement) + return + + raise PolicyViolation(f"Unsupported non-deleting operation: {operation}") + + +def apply_static_repair( + record: dict[str, Any], + repair_plan: dict[str, Any], +) -> tuple[dict[str, Any], dict[str, Any]]: + """Apply an approved allowlisted plan and return a structured audit diff. + + The function performs only syntactic operations. It never executes code and + never changes outcome labels or patch metadata. + """ + + enforce_repair_policy(repair_plan) + decision = repair_plan["repair_decision"] + if decision not in {"APPLY_STATIC_REPAIR", "CREATE_STEP_EXAMPLE"}: + raise PolicyViolation(f"Repair decision {decision} does not authorize mutation") + + original = copy.deepcopy(record) + repaired = copy.deepcopy(record) + entries = [ + {"original_turn_id": turn_id, "message": copy.deepcopy(message)} + for turn_id, message in enumerate(repaired.get("trajectory") or [], 1) + ] + removed_turns: list[int] = [] + replaced_turns: list[int] = [] + + def find_entry(turn_id: int) -> dict[str, Any]: + for item in entries: + if item["original_turn_id"] == turn_id: + return item + raise PolicyViolation(f"Repair plan references missing or already removed turn {turn_id}") + + for operation in repair_plan["operations"]: + op = operation["op"] + targets = operation["target_turns"] + replacement = operation["replacement"] + + if op == "TRUNCATE_BEFORE_TURN": + if len(targets) != 1: + raise PolicyViolation("TRUNCATE_BEFORE_TURN requires exactly one target turn") + cutoff = targets[0] + to_remove = [item["original_turn_id"] for item in entries if item["original_turn_id"] >= cutoff] + entries[:] = [item for item in entries if item["original_turn_id"] < cutoff] + removed_turns.extend(to_remove) + continue + + if op == "DROP_REDUNDANT_READ_ONLY_PAIR": + if len(targets) != 2 or targets[1] != targets[0] + 1: + raise PolicyViolation("A read-only pair must contain two consecutive original turns") + assistant_entry = find_entry(targets[0]) + tool_entry = find_entry(targets[1]) + if not _is_read_only_pair(assistant_entry["message"], tool_entry["message"]): + raise PolicyViolation("The requested pair is not provably read-only") + entries[:] = [item for item in entries if item["original_turn_id"] not in set(targets)] + removed_turns.extend(targets) + continue + + if len(targets) != 1: + raise PolicyViolation(f"{op} requires exactly one target turn") + entry = find_entry(targets[0]) + _replace_message(entry, op, replacement) + replaced_turns.append(targets[0]) + + repaired["trajectory"] = [item["message"] for item in entries] + + # Deterministic invariants protect immutable outcome and patch metadata. + if repaired.get("resolved") != original.get("resolved"): + raise PolicyViolation("Static repair changed resolved") + original_metadata = original.get("metadata") or {} + repaired_metadata = repaired.get("metadata") or {} + for key in ("model_patch", "reference_patch"): + if repaired_metadata.get(key) != original_metadata.get(key): + raise PolicyViolation(f"Static repair changed immutable metadata field {key}") + + # Every retained tool observation must remain byte-for-byte equivalent. + original_tool_hashes = { + turn_id: _hash(message) + for turn_id, message in enumerate(original.get("trajectory") or [], 1) + if isinstance(message, dict) and message.get("role") == "tool" + } + retained_entries = {item["original_turn_id"]: item["message"] for item in entries} + for turn_id, digest in original_tool_hashes.items(): + if turn_id in retained_entries and _hash(retained_entries[turn_id]) != digest: + raise PolicyViolation(f"Static repair modified retained tool output at turn {turn_id}") + + diff = { + "sample_id": get_sample_id(record), + "input_sha256": _hash(original), + "output_sha256": _hash(repaired), + "removed_original_turn_ids": sorted(set(removed_turns)), + "replaced_original_turn_ids": sorted(set(replaced_turns)), + "original_turn_count": len(original.get("trajectory") or []), + "repaired_turn_count": len(repaired.get("trajectory") or []), + "maximum_training_use": repair_plan["maximum_training_use"], + } + return repaired, diff diff --git a/src/swe_data_processing/resources.py b/src/swe_data_processing/resources.py new file mode 100644 index 0000000..883d0ec --- /dev/null +++ b/src/swe_data_processing/resources.py @@ -0,0 +1,21 @@ +"""Load packaged prompts and JSON Schemas by stable resource name.""" + +from __future__ import annotations + +import json +from importlib.resources import files +from typing import Any + + +def load_prompt(name: str) -> str: + """Load a UTF-8 prompt from the package's ``prompts`` directory.""" + + resource = files("swe_data_processing").joinpath("prompts", name) + return resource.read_text(encoding="utf-8") + + +def load_schema(name: str) -> dict[str, Any]: + """Load and decode a JSON Schema from packaged resources.""" + + resource = files("swe_data_processing").joinpath("schemas", name) + return json.loads(resource.read_text(encoding="utf-8")) diff --git a/src/swe_data_processing/schemas/classification_output.schema.json b/src/swe_data_processing/schemas/classification_output.schema.json new file mode 100644 index 0000000..4069b25 --- /dev/null +++ b/src/swe_data_processing/schemas/classification_output.schema.json @@ -0,0 +1,128 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "additionalProperties": false, + "required": [ + "sample_id", "source_outcome_class", "qc_decision", "training_use", "qc_passed", + "confidence", "verification", "dimensions", "hard_fail_codes", "warning_codes", + "evidence", "repairability", "summary" + ], + "properties": { + "sample_id": {"type": "string", "minLength": 1}, + "source_outcome_class": { + "enum": ["POSITIVE_CANDIDATE", "EXPLICIT_NEGATIVE", "UNVERIFIED"] + }, + "qc_decision": { + "enum": [ + "ACCEPT_SILVER_POSITIVE", "ACCEPT_NEGATIVE", "STATIC_REPAIR", + "HOLD_UNVERIFIED", "REJECT" + ] + }, + "training_use": { + "enum": ["SFT_FULL", "SFT_STEP_ONLY", "DPO_REJECTED", "ERROR_ANALYSIS", "HOLD", "DROP"] + }, + "qc_passed": {"type": "boolean"}, + "confidence": {"type": "number", "minimum": 0, "maximum": 1}, + "verification": { + "type": "object", + "additionalProperties": false, + "required": [ + "status", "level", "last_code_change_turn", "last_relevant_test_turn", + "test_after_final_change", "scope_relevant", "exit_code_trustworthy" + ], + "properties": { + "status": { + "enum": ["PASS_RELIABLE", "FAIL_EXPLICIT", "MASKED_EXIT", "ENV_BLOCKED", "INSUFFICIENT", "NONE"] + }, + "level": {"enum": ["NONE", "SYNTAX", "BUILD", "TARGETED", "MODULE", "FULL"]}, + "last_code_change_turn": {"type": ["integer", "null"], "minimum": 1}, + "last_relevant_test_turn": {"type": ["integer", "null"], "minimum": 1}, + "test_after_final_change": {"type": "boolean"}, + "scope_relevant": {"type": "boolean"}, + "exit_code_trustworthy": {"type": "boolean"} + } + }, + "dimensions": { + "type": "object", + "additionalProperties": false, + "required": [ + "trajectory_integrity", "tool_integrity", "patch_presence", + "patch_trajectory_consistency", "instruction_compliance", "verification_consistency", + "final_claim_alignment", "patch_hygiene", "issue_patch_alignment" + ], + "properties": { + "trajectory_integrity": {"$ref": "#/$defs/dimension"}, + "tool_integrity": {"$ref": "#/$defs/dimension"}, + "patch_presence": {"$ref": "#/$defs/dimension"}, + "patch_trajectory_consistency": {"$ref": "#/$defs/dimension"}, + "instruction_compliance": {"$ref": "#/$defs/dimension"}, + "verification_consistency": {"$ref": "#/$defs/dimension"}, + "final_claim_alignment": {"$ref": "#/$defs/dimension"}, + "patch_hygiene": {"$ref": "#/$defs/dimension"}, + "issue_patch_alignment": {"$ref": "#/$defs/dimension"} + } + }, + "hard_fail_codes": { + "type": "array", "uniqueItems": true, + "items": {"$ref": "#/$defs/hardFailCode"} + }, + "warning_codes": { + "type": "array", "uniqueItems": true, + "items": {"$ref": "#/$defs/warningCode"} + }, + "evidence": { + "type": "array", "minItems": 1, "maxItems": 20, + "items": { + "type": "object", "additionalProperties": false, + "required": ["turn_id", "field", "code", "quote"], + "properties": { + "turn_id": {"type": ["integer", "null"], "minimum": 1}, + "field": {"type": "string", "minLength": 1, "maxLength": 80}, + "code": {"type": "string", "minLength": 1, "maxLength": 80}, + "quote": {"type": "string", "minLength": 1, "maxLength": 240} + } + } + }, + "repairability": { + "type": "object", "additionalProperties": false, + "required": ["decision", "safe_operations"], + "properties": { + "decision": { + "enum": ["NOT_NEEDED", "SAFE_STATIC", "STEP_ONLY", "REQUIRES_EXECUTION", "NOT_REPAIRABLE"] + }, + "safe_operations": { + "type": "array", "uniqueItems": true, + "items": {"$ref": "#/$defs/repairOperation"} + } + } + }, + "summary": {"type": "string", "minLength": 1, "maxLength": 1200} + }, + "$defs": { + "dimension": {"enum": ["PASS", "FAIL", "UNKNOWN"]}, + "hardFailCode": { + "enum": [ + "RESOLVED_ZERO_FOR_POSITIVE", "FINAL_RELEVANT_TEST_FAILED", "SUCCESS_LOG_CONTRADICTION", + "MALFORMED_TOOL_ARGUMENTS", "UNKNOWN_TOOL", "UNPAIRED_TOOL_RESULT", "MISSING_PATCH", + "PATCH_TRAJECTORY_MISMATCH", "INSTRUCTION_VIOLATION", "PATCH_CONTAMINATION_SEVERE", + "INCOMPLETE_TRAJECTORY", "UNSAFE_OR_UNAUTHORIZED_ACTION", "ISSUE_PATCH_MISMATCH", + "POST_TEST_CODE_CHANGE", "UNPROVEN_PREEXISTING_FAILURE" + ] + }, + "warningCode": { + "enum": [ + "UNTRUSTWORTHY_TEST_PIPELINE", "NO_VERIFICATION", "BUILD_ONLY", "TARGETED_TEST_ONLY", + "ENVIRONMENT_FAILURE", "REFERENCE_PATCH_LOW_OVERLAP", "PATCH_SIZE_OUTLIER", + "REDUNDANT_TOOL_CALLS", "REPETITIVE_SUCCESS_CLAIMS", "TEMPORARY_FILES_OBSERVED", + "TEST_FILE_MODIFIED", "VENDOR_OR_GENERATED_FILE_MODIFIED" + ] + }, + "repairOperation": { + "enum": [ + "CANONICALIZE_JSON", "NORMALIZE_TOOL_ALIAS", "REDACT_SECRET", "REMOVE_DUPLICATE_PROSE", + "REWRITE_FINAL_SUMMARY", "DROP_REDUNDANT_READ_ONLY_PAIR", "TRUNCATE_BEFORE_TURN", + "EMIT_STRUCTURALLY_CORRECT_ASSISTANT_TURN" + ] + } + } +} diff --git a/src/swe_data_processing/schemas/repair_output.schema.json b/src/swe_data_processing/schemas/repair_output.schema.json new file mode 100644 index 0000000..7caa70b --- /dev/null +++ b/src/swe_data_processing/schemas/repair_output.schema.json @@ -0,0 +1,63 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "additionalProperties": false, + "required": [ + "sample_id", "repair_decision", "maximum_training_use", "invariants", + "operations", "requires_second_review", "summary" + ], + "properties": { + "sample_id": {"type": "string", "minLength": 1}, + "repair_decision": { + "enum": ["NO_CHANGE", "APPLY_STATIC_REPAIR", "CREATE_STEP_EXAMPLE", "REQUIRES_EXECUTION", "DROP"] + }, + "maximum_training_use": { + "enum": ["SFT_FULL", "SFT_STEP_ONLY", "DPO_REJECTED", "ERROR_ANALYSIS", "HOLD", "DROP"] + }, + "invariants": { + "type": "object", "additionalProperties": false, + "required": [ + "resolved_unchanged", "tool_outputs_unchanged", "model_patch_unchanged", + "reference_patch_unchanged", "no_synthetic_execution_result" + ], + "properties": { + "resolved_unchanged": {"const": true}, + "tool_outputs_unchanged": {"const": true}, + "model_patch_unchanged": {"const": true}, + "reference_patch_unchanged": {"const": true}, + "no_synthetic_execution_result": {"const": true} + } + }, + "operations": { + "type": "array", "maxItems": 30, + "items": { + "type": "object", "additionalProperties": false, + "required": ["op", "target_turns", "preconditions", "replacement", "reason"], + "properties": { + "op": {"$ref": "#/$defs/repairOperation"}, + "target_turns": { + "type": "array", "uniqueItems": true, + "items": {"type": "integer", "minimum": 1} + }, + "preconditions": { + "type": "array", "minItems": 1, + "items": {"type": "string", "minLength": 1, "maxLength": 240} + }, + "replacement": {"type": ["string", "object", "array", "null"]}, + "reason": {"type": "string", "minLength": 1, "maxLength": 600} + } + } + }, + "requires_second_review": {"const": true}, + "summary": {"type": "string", "minLength": 1, "maxLength": 1200} + }, + "$defs": { + "repairOperation": { + "enum": [ + "CANONICALIZE_JSON", "NORMALIZE_TOOL_ALIAS", "REDACT_SECRET", "REMOVE_DUPLICATE_PROSE", + "REWRITE_FINAL_SUMMARY", "DROP_REDUNDANT_READ_ONLY_PAIR", "TRUNCATE_BEFORE_TURN", + "EMIT_STRUCTURALLY_CORRECT_ASSISTANT_TURN" + ] + } + } +} diff --git a/src/swe_data_processing/schemas/review_output.schema.json b/src/swe_data_processing/schemas/review_output.schema.json new file mode 100644 index 0000000..9254ba5 --- /dev/null +++ b/src/swe_data_processing/schemas/review_output.schema.json @@ -0,0 +1,43 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "additionalProperties": false, + "required": [ + "sample_id", "approved", "maximum_training_use", "invariant_violations", + "unsupported_changes", "evidence", "summary" + ], + "properties": { + "sample_id": {"type": "string", "minLength": 1}, + "approved": {"type": "boolean"}, + "maximum_training_use": { + "enum": ["SFT_FULL", "SFT_STEP_ONLY", "DPO_REJECTED", "ERROR_ANALYSIS", "HOLD", "DROP"] + }, + "invariant_violations": { + "type": "array", "uniqueItems": true, + "items": { + "enum": [ + "RESOLVED_CHANGED", "TOOL_OUTPUT_CHANGED", "MODEL_PATCH_CHANGED", "REFERENCE_PATCH_CHANGED", + "SYNTHETIC_EXECUTION_RESULT_ADDED", "STATEFUL_TURN_REMOVED", "DANGLING_TOOL_PAIR", + "SEMANTIC_TOOL_CALL_CHANGED_WITHOUT_EXECUTION", "TRAINING_USE_UPGRADED_WITHOUT_EVIDENCE" + ] + } + }, + "unsupported_changes": { + "type": "array", + "items": {"type": "string", "minLength": 1, "maxLength": 400} + }, + "evidence": { + "type": "array", "minItems": 1, "maxItems": 20, + "items": { + "type": "object", "additionalProperties": false, + "required": ["turn_id", "quote", "assessment"], + "properties": { + "turn_id": {"type": ["integer", "null"], "minimum": 1}, + "quote": {"type": "string", "minLength": 1, "maxLength": 240}, + "assessment": {"type": "string", "minLength": 1, "maxLength": 400} + } + } + }, + "summary": {"type": "string", "minLength": 1, "maxLength": 1000} + } +} diff --git a/src/swe_data_processing/schemas/trajectory_audit_output.schema.json b/src/swe_data_processing/schemas/trajectory_audit_output.schema.json new file mode 100644 index 0000000..20e42ec --- /dev/null +++ b/src/swe_data_processing/schemas/trajectory_audit_output.schema.json @@ -0,0 +1,116 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "additionalProperties": false, + "required": [ + "sample_id", "evaluation_mode", "truncation", "behavior_issues", "issue_counts", + "quality_dimensions", "recommended_use", "confidence", "summary" + ], + "properties": { + "sample_id": {"type": "string", "minLength": 1}, + "evaluation_mode": {"enum": ["SUCCESS_QUALITY", "PROCESS_SALVAGE"]}, + "truncation": { + "type": "object", + "additionalProperties": false, + "required": [ + "first_bad_assistant_turn", "truncate_before_turn", "acceptable_start_turn", + "acceptable_end_turn", "prefix_usable", "category", "evidence", "reason" + ], + "properties": { + "first_bad_assistant_turn": {"type": ["integer", "null"], "minimum": 1}, + "truncate_before_turn": {"type": ["integer", "null"], "minimum": 1}, + "acceptable_start_turn": {"type": ["integer", "null"], "minimum": 1}, + "acceptable_end_turn": {"type": ["integer", "null"], "minimum": 1}, + "prefix_usable": {"type": "boolean"}, + "category": { + "enum": [ + "NONE", "IGNORED_FAILURE", "CONSTRAINT_VIOLATION", "UNGROUNDED_SUCCESS", + "WRONG_TOOL_OR_TARGET", "UNSAFE_STATE_CHANGE", "REPEATED_NO_PROGRESS", + "MISREAD_OBSERVATION", "INVALID_TOOL_CALL" + ] + }, + "evidence": { + "type": "array", "maxItems": 4, + "items": {"$ref": "#/$defs/evidence"} + }, + "reason": {"type": "string", "minLength": 1, "maxLength": 1400} + } + }, + "behavior_issues": { + "type": "array", "maxItems": 40, + "items": { + "type": "object", + "additionalProperties": false, + "required": [ + "assistant_turn", "tool_result_turn", "tool_name", "kind", "category", + "severity", "state_effect", "recovered_later", "quote", "reason" + ], + "properties": { + "assistant_turn": {"type": "integer", "minimum": 1}, + "tool_result_turn": {"type": ["integer", "null"], "minimum": 1}, + "tool_name": {"type": ["string", "null"], "maxLength": 80}, + "kind": {"enum": ["ERROR", "INEFFICIENCY"]}, + "category": { + "enum": [ + "IGNORED_FAILURE", "CONSTRAINT_VIOLATION", "UNGROUNDED_CLAIM", + "WRONG_TOOL_OR_TARGET", "MISREAD_OBSERVATION", "INVALID_ARGUMENT", "INVALID_TOOL_CALL", + "UNSAFE_ACTION", + "REDUNDANT_READ", "REPEATED_FAILURE", "NO_PROGRESS_LOOP", + "OVER_VERIFICATION", "REPETITIVE_SUMMARY", "MASKED_EXIT_STATUS", + "ENVIRONMENT_MISDIAGNOSIS", "DISCARDED_USEFUL_EVIDENCE" + ] + }, + "severity": {"enum": ["MINOR", "MAJOR", "CRITICAL"]}, + "state_effect": {"enum": ["NONE", "POSSIBLE", "CONFIRMED"]}, + "recovered_later": {"type": "boolean"}, + "quote": {"type": "string", "minLength": 1, "maxLength": 800}, + "reason": {"type": "string", "minLength": 1, "maxLength": 500} + } + } + }, + "issue_counts": { + "type": "object", + "additionalProperties": false, + "required": ["errors", "inefficiencies", "critical", "major", "minor"], + "properties": { + "errors": {"type": "integer", "minimum": 0}, + "inefficiencies": {"type": "integer", "minimum": 0}, + "critical": {"type": "integer", "minimum": 0}, + "major": {"type": "integer", "minimum": 0}, + "minor": {"type": "integer", "minimum": 0} + } + }, + "quality_dimensions": { + "type": "object", + "additionalProperties": false, + "required": [ + "planning", "tool_selection", "observation_use", "efficiency", + "verification_discipline", "claim_calibration" + ], + "properties": { + "planning": {"type": "integer", "minimum": 0, "maximum": 5}, + "tool_selection": {"type": "integer", "minimum": 0, "maximum": 5}, + "observation_use": {"type": "integer", "minimum": 0, "maximum": 5}, + "efficiency": {"type": "integer", "minimum": 0, "maximum": 5}, + "verification_discipline": {"type": "integer", "minimum": 0, "maximum": 5}, + "claim_calibration": {"type": "integer", "minimum": 0, "maximum": 5} + } + }, + "recommended_use": { + "enum": ["FULL_TRAJECTORY_CANDIDATE", "PROCESS_PREFIX_CANDIDATE", "HOLD", "REJECT"] + }, + "confidence": {"type": "number", "minimum": 0, "maximum": 1}, + "summary": {"type": "string", "minLength": 1, "maxLength": 1000} + }, + "$defs": { + "evidence": { + "type": "object", + "additionalProperties": false, + "required": ["turn_id", "quote"], + "properties": { + "turn_id": {"type": "integer", "minimum": 1}, + "quote": {"type": "string", "minLength": 1, "maxLength": 800} + } + } + } +} diff --git a/src/swe_data_processing/workflow.py b/src/swe_data_processing/workflow.py new file mode 100644 index 0000000..12991be --- /dev/null +++ b/src/swe_data_processing/workflow.py @@ -0,0 +1,202 @@ +"""High-level classification, repair planning, and review workflows.""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from typing import Any + +from .audit import compute_quality_score, validate_audit +from .client import GLMClient, GLMResponse +from .evidence import compact_patch_object, compact_text, compact_trajectory +from .features import extract_instruction_constraints, extract_static_signals, index_trajectory +from .io import get_sample_id +from .policy import enforce_classification_policy, enforce_repair_policy +from .resources import load_prompt, load_schema + + +def _stable_hash(value: dict[str, Any]) -> str: + """Hash canonical JSON for reproducible provenance records.""" + + encoded = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _first_user_issue(trajectory: list[dict[str, Any]]) -> str: + """Return the first user message, which contains the benchmark issue.""" + + for message in trajectory: + if message.get("role") == "user": + content = message.get("content") + if isinstance(content, str): + return content + return "" + + +def prepare_classification_payload(record: dict[str, Any]) -> dict[str, Any]: + """Create the complete evidence package consumed by the classifier.""" + + trajectory = index_trajectory(record) + metadata = record.get("metadata") or {} + static_signals = extract_static_signals(record) + trajectory_evidence, compaction = compact_trajectory(trajectory, static_signals) + return { + "sample_id": get_sample_id(record), + "resolved": int(record.get("resolved", -1)), + "issue": compact_text(_first_user_issue(trajectory), 16_000), + "instruction_constraints": extract_instruction_constraints(trajectory), + "tool_definition_summary": { + "allowed_tool_names": static_signals["allowed_tool_names"], + "malformed_definition_indexes": static_signals[ + "malformed_tool_definition_indexes" + ], + }, + "trajectory": trajectory_evidence, + "prompt_compaction": compaction, + "model_patch": compact_patch_object(metadata.get("model_patch") or {}), + "reference_patch": compact_patch_object(metadata.get("reference_patch") or {}), + "static_signals": static_signals, + "record_metadata": { + "instance_id": record.get("instance_id"), + "repo": record.get("repo"), + "language": record.get("language"), + "trajectory_id": record.get("trajectory_id"), + "category": metadata.get("category") if isinstance(metadata, dict) else None, + }, + } + + +def _provenance( + stage: str, + response: GLMResponse, + input_payload: dict[str, Any], + client: GLMClient, +) -> dict[str, Any]: + """Build non-sensitive provenance for one successful API stage.""" + + return { + "stage": stage, + "model": client.settings.model, + "endpoint": client.settings.endpoint, + "request_id": response.request_id, + "usage": response.usage, + "compatibility_fallback_used": response.compatibility_fallback_used, + "input_sha256": _stable_hash(input_payload), + "created_at": datetime.now(timezone.utc).isoformat(), + } + + +def classify_record(record: dict[str, Any], client: GLMClient) -> dict[str, Any]: + """Classify one trajectory and enforce immutable outcome policy locally.""" + + payload = prepare_classification_payload(record) + response = client.invoke_json( + system_prompt=load_prompt("classification.md"), + payload=payload, + schema=load_schema("classification_output.schema.json"), + ) + enforce_classification_policy(record, response.data) + result = dict(response.data) + result["provenance"] = _provenance("classification", response, payload, client) + return result + + +def prepare_audit_payload(record: dict[str, Any]) -> dict[str, Any]: + """Build an audit payload with all reference-solution evidence removed.""" + + payload = prepare_classification_payload(record) + # The reference solution is deliberately hidden from this task. A causal + # bad-turn label must be supported by information the original agent had. + # Reference-derived summary features must also be removed; otherwise file + # names and size ratios can leak the hidden solution even without its text. + payload.pop("reference_patch", None) + static_signals = dict(payload["static_signals"]) + for key in ( + "reference_patch_files", + "reference_patch_chars", + "patch_size_ratio_to_reference", + ): + static_signals.pop(key, None) + payload["static_signals"] = static_signals + resolved = int(record.get("resolved", -1)) + payload["evaluation_mode"] = "SUCCESS_QUALITY" if resolved == 1 else "PROCESS_SALVAGE" + payload["outcome_label_usage"] = ( + "Routing only. Do not use resolved to identify a bad action or justify a truncation turn." + ) + return payload + + +def audit_trajectory(record: dict[str, Any], client: GLMClient) -> dict[str, Any]: + """Audit causal process quality and possible prefix salvage without repair.""" + + payload = prepare_audit_payload(record) + response = client.invoke_json( + system_prompt=load_prompt("trajectory_audit.md"), + payload=payload, + schema=load_schema("trajectory_audit_output.schema.json"), + ) + validate_audit(record, response.data) + result = dict(response.data) + result["source_record"] = { + "resolved": record.get("resolved"), + "instance_id": record.get("instance_id"), + "repo": record.get("repo"), + "language": record.get("language"), + "sample_provenance": record.get("_sample"), + } + result["local_quality_score"] = compute_quality_score(result) + result["provenance"] = _provenance("trajectory_audit", response, payload, client) + return result + + +def plan_static_repair( + record: dict[str, Any], + classification: dict[str, Any], + client: GLMClient, +) -> dict[str, Any]: + """Produce a fact-preserving repair plan without applying changes.""" + + payload = { + "sample_id": get_sample_id(record), + "original_record": prepare_classification_payload(record), + "classification": classification, + } + response = client.invoke_json( + system_prompt=load_prompt("repair.md"), + payload=payload, + schema=load_schema("repair_output.schema.json"), + ) + enforce_repair_policy(response.data) + result = dict(response.data) + result["provenance"] = _provenance("repair_plan", response, payload, client) + return result + + +def review_repaired_record( + *, + original_record: dict[str, Any], + classification: dict[str, Any], + repair_plan: dict[str, Any], + repaired_record: dict[str, Any], + structured_diff: dict[str, Any], + client: GLMClient, +) -> dict[str, Any]: + """Independently review a repaired record against immutable evidence.""" + + payload = { + "sample_id": get_sample_id(original_record), + "original_record": original_record, + "classification": classification, + "repair_plan": repair_plan, + "repaired_record": repaired_record, + "structured_diff": structured_diff, + } + response = client.invoke_json( + system_prompt=load_prompt("review.md"), + payload=payload, + schema=load_schema("review_output.schema.json"), + ) + result = dict(response.data) + result["provenance"] = _provenance("repair_review", response, payload, client) + return result diff --git a/tests/test_audit.py b/tests/test_audit.py new file mode 100644 index 0000000..acc030b --- /dev/null +++ b/tests/test_audit.py @@ -0,0 +1,114 @@ +"""Tests for local trajectory-audit policy and scoring.""" + +from __future__ import annotations + +import pytest + +from swe_data_processing.audit import compute_quality_score, validate_audit +from swe_data_processing.policy import PolicyViolation +from swe_data_processing.workflow import prepare_audit_payload + + +def _audit_result() -> dict: + """Return one internally consistent audit result.""" + + return { + "evaluation_mode": "PROCESS_SALVAGE", + "recommended_use": "PROCESS_PREFIX_CANDIDATE", + "truncation": { + "first_bad_assistant_turn": 3, + "truncate_before_turn": 3, + "acceptable_start_turn": 3, + "acceptable_end_turn": 3, + "prefix_usable": True, + }, + "behavior_issues": [ + { + "assistant_turn": 3, + "tool_result_turn": 4, + "kind": "ERROR", + "severity": "MAJOR", + } + ], + "issue_counts": { + "errors": 1, + "inefficiencies": 0, + "critical": 0, + "major": 1, + "minor": 0, + }, + "quality_dimensions": { + "planning": 4, + "tool_selection": 4, + "observation_use": 3, + "efficiency": 4, + "verification_discipline": 3, + "claim_calibration": 3, + }, + } + + +def _record() -> dict: + return { + "trajectory": [ + {"role": "user", "content": "issue"}, + {"role": "tool", "content": "context"}, + {"role": "assistant", "content": "bad call"}, + {"role": "tool", "content": "failed"}, + ] + } + + +def test_valid_audit_and_score() -> None: + """A consistent audit receives a deterministic bounded score.""" + + result = _audit_result() + validate_audit(_record(), result) + score = compute_quality_score(result) + assert score["educational_quality_score"] == 65.0 + assert score["quality_tier"] == "LOW" + + +def test_first_bad_turn_must_be_assistant() -> None: + """The truncation boundary cannot point at a tool observation.""" + + result = _audit_result() + result["truncation"]["first_bad_assistant_turn"] = 4 + result["truncation"]["truncate_before_turn"] = 4 + with pytest.raises(PolicyViolation, match="assistant turn"): + validate_audit(_record(), result) + + +def test_issue_counts_are_recomputed() -> None: + """GLM cannot under-report the number of issues it listed.""" + + result = _audit_result() + result["issue_counts"]["errors"] = 0 + with pytest.raises(PolicyViolation, match="issue_counts"): + validate_audit(_record(), result) + + +def test_process_salvage_cannot_become_full_trajectory_candidate() -> None: + """Outcome routing cannot be overridden by the model recommendation.""" + + result = _audit_result() + result["recommended_use"] = "FULL_TRAJECTORY_CANDIDATE" + with pytest.raises(PolicyViolation, match="full-trajectory"): + validate_audit(_record(), result) + + +def test_audit_payload_removes_reference_solution_signals() -> None: + """The causal audit cannot see reference patch text or derived metadata.""" + + record = { + "trajectory_id": "sample-1", + "resolved": 0, + "trajectory": [{"role": "user", "content": "Fix the issue"}], + "metadata": { + "model_patch": {"patch": "--- a/model.py\n+++ b/model.py\n"}, + "reference_patch": {"patch": "--- a/secret.py\n+++ b/secret.py\n"}, + }, + } + payload = prepare_audit_payload(record) + assert "reference_patch" not in payload + assert not any("reference" in key for key in payload["static_signals"]) diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..7fe318b --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,71 @@ +"""Tests for API request compatibility and structured-response validation.""" + +from __future__ import annotations + +import json + +import httpx + +from swe_data_processing.client import GLMClient +from swe_data_processing.config import Settings + + +def test_client_parses_fenced_json_and_never_sends_key_in_body() -> None: + captured_body = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.update(json.loads(request.content)) + assert request.headers["authorization"] == "Bearer test-secret" + return httpx.Response( + 200, + headers={"x-request-id": "request-123"}, + json={ + "choices": [ + {"message": {"role": "assistant", "content": "```json\n{\"status\":\"ok\"}\n```"}} + ], + "usage": {"total_tokens": 10}, + }, + ) + + schema = { + "type": "object", + "additionalProperties": False, + "required": ["status"], + "properties": {"status": {"const": "ok"}}, + } + settings = Settings(api_key="test-secret", max_retries=1) + with GLMClient(settings, transport=httpx.MockTransport(handler)) as client: + response = client.invoke_json(system_prompt="Return JSON.", payload={"ping": True}, schema=schema) + + assert response.data == {"status": "ok"} + assert response.request_id == "request-123" + assert response.usage == {"total_tokens": 10} + assert "test-secret" not in json.dumps(captured_body) + assert captured_body["model"] == "glm-5.2" + + +def test_client_falls_back_when_vendor_extensions_are_rejected() -> None: + calls = [] + + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + calls.append(body) + if len(calls) == 1: + return httpx.Response(400, json={"error": {"message": "unknown parameter: thinking"}}) + return httpx.Response( + 200, + json={"choices": [{"message": {"content": "{\"status\":\"ok\"}"}}]}, + ) + + schema = { + "type": "object", + "required": ["status"], + "properties": {"status": {"const": "ok"}}, + } + settings = Settings(api_key="test-secret", max_retries=2) + with GLMClient(settings, transport=httpx.MockTransport(handler)) as client: + response = client.invoke_json(system_prompt="Return JSON.", payload={}, schema=schema) + + assert "thinking" in calls[0] + assert "thinking" not in calls[1] + assert response.compatibility_fallback_used is True diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..e9b054d --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,27 @@ +"""Tests for strict environment configuration and endpoint normalization.""" + +from __future__ import annotations + +import pytest + +from swe_data_processing.config import Settings + + +def test_endpoint_normalization() -> None: + settings = Settings( + api_key="test-secret", + api_base="https://llm-api.cowin.run/", + api_path="v1/chat/completions", + ) + assert settings.endpoint == "https://llm-api.cowin.run/v1/chat/completions" + + +def test_missing_api_key_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("GLM_API_KEY", raising=False) + with pytest.raises(ValueError, match="GLM_API_KEY"): + Settings.from_env() + + +def test_feature_only_settings_can_omit_key(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("GLM_API_KEY", raising=False) + assert Settings.from_env(require_api_key=False).api_key == "" diff --git a/tests/test_evidence.py b/tests/test_evidence.py new file mode 100644 index 0000000..dc336b7 --- /dev/null +++ b/tests/test_evidence.py @@ -0,0 +1,38 @@ +"""Tests for prompt-only trajectory compaction.""" + +from swe_data_processing.evidence import compact_text, compact_trajectory + + +def test_compact_text_preserves_short_values() -> None: + """Short evidence must remain byte-for-byte identical.""" + + assert compact_text("short evidence", 100) == "short evidence" + + +def test_compact_text_hashes_long_values() -> None: + """Long evidence must expose its size and digest for auditability.""" + + result = compact_text("a" * 1_000, 200) + assert len(result) == 200 + assert "COMPACTED original_chars=1000 sha256=" in result + + +def test_compact_trajectory_prioritizes_test_observations() -> None: + """Important test output receives a larger preview than ordinary output.""" + + long_content = "x" * 3_000 + trajectory = [ + {"turn_id": 1, "role": "user", "content": "issue"}, + {"turn_id": 2, "role": "assistant", "content": "", "tool_calls": []}, + {"turn_id": 3, "role": "tool", "content": long_content}, + {"turn_id": 4, "role": "assistant", "content": "done"}, + ] + signals = { + "stateful_turns": [], + "test_events": [{"command_turn": 2, "output_turn": 3}], + "malformed_tool_turns": [], + "unknown_tool_turns": [], + } + compacted, metadata = compact_trajectory(trajectory, signals) + assert compacted[2]["content"] == long_content + assert 3 in metadata["important_turn_ids"] diff --git a/tests/test_features.py b/tests/test_features.py new file mode 100644 index 0000000..b844fb9 --- /dev/null +++ b/tests/test_features.py @@ -0,0 +1,100 @@ +"""Tests for deterministic static trajectory evidence extraction.""" + +from __future__ import annotations + +import json + +from swe_data_processing.features import extract_patch_files, extract_static_signals + + +def _tool(name: str) -> str: + return json.dumps({"type": "function", "function": {"name": name, "parameters": {}}}) + + +def test_detects_masked_failing_test_after_edit() -> None: + record = { + "trajectory_id": "sample-1", + "resolved": 0, + "tools": [_tool("str_replace_editor"), _tool("execute_bash"), _tool("finish")], + "trajectory": [ + {"role": "system", "content": "system"}, + {"role": "user", "content": "Do not modify test files."}, + { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "edit-1", + "type": "function", + "function": { + "name": "str_replace_editor", + "arguments": json.dumps( + {"command": "str_replace", "path": "/workspace/src/main.py"} + ), + }, + } + ], + }, + {"role": "tool", "content": "The file was edited."}, + { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "test-1", + "type": "function", + "function": { + "name": "execute_bash", + "arguments": json.dumps({"command": "pytest tests | tail -20"}), + }, + } + ], + }, + { + "role": "tool", + "content": "1 failed, 10 passed\n[Command finished with exit code 0]", + }, + ], + "metadata": { + "model_patch": {"patch": "diff --git a/src/main.py b/src/main.py\n+++ b/src/main.py\n"}, + "reference_patch": {"patch": "diff --git a/src/main.py b/src/main.py\n+++ b/src/main.py\n"}, + }, + } + + signals = extract_static_signals(record) + assert signals["last_stateful_turn"] == 3 + assert signals["instruction_constraints"] == ["Do not modify test files."] + assert signals["test_events"][0]["masked_pipeline"] is True + assert signals["test_events"][0]["explicit_failure"] is True + assert signals["test_events"][0]["explicit_exit_codes"] == [0] + + +def test_detects_malformed_and_unknown_tool_calls() -> None: + record = { + "trajectory_id": "sample-2", + "tools": [_tool("execute_bash")], + "trajectory": [ + {"role": "system", "content": "system"}, + {"role": "user", "content": "issue"}, + { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "bad", + "function": {"name": "made_up_tool", "arguments": "{bad json"}, + } + ], + }, + {"role": "tool", "content": "error"}, + ], + "metadata": {"model_patch": {}, "reference_patch": {}}, + } + signals = extract_static_signals(record) + assert signals["malformed_tool_turns"][0]["turn_id"] == 3 + assert signals["unknown_tool_turns"] == [{"turn_id": 3, "tool_name": "made_up_tool"}] + + +def test_extract_patch_files_is_stable_and_unique() -> None: + patch = "+++ b/src/a.py\n+++ b/src/b.py\n+++ b/src/a.py\n+++ /dev/null\n" + assert extract_patch_files(patch) == ["src/a.py", "src/b.py"] diff --git a/tests/test_policy.py b/tests/test_policy.py new file mode 100644 index 0000000..c3624f1 --- /dev/null +++ b/tests/test_policy.py @@ -0,0 +1,100 @@ +"""Tests for immutable classification and repair policy enforcement.""" + +from __future__ import annotations + +import pytest + +from swe_data_processing.policy import ( + PolicyViolation, + enforce_classification_policy, + enforce_repair_policy, +) + + +def _classification() -> dict: + dimensions = { + "trajectory_integrity": "PASS", + "tool_integrity": "PASS", + "patch_presence": "PASS", + "patch_trajectory_consistency": "PASS", + "instruction_compliance": "PASS", + "verification_consistency": "PASS", + "final_claim_alignment": "PASS", + "patch_hygiene": "PASS", + "issue_patch_alignment": "PASS", + } + return { + "source_outcome_class": "POSITIVE_CANDIDATE", + "qc_decision": "ACCEPT_SILVER_POSITIVE", + "training_use": "SFT_FULL", + "qc_passed": True, + "hard_fail_codes": [], + "verification": {"status": "PASS_RELIABLE"}, + "dimensions": dimensions, + } + + +def test_valid_silver_positive_passes() -> None: + enforce_classification_policy({"resolved": 1}, _classification()) + + +def test_resolved_zero_cannot_be_promoted() -> None: + result = _classification() + result["source_outcome_class"] = "EXPLICIT_NEGATIVE" + with pytest.raises(PolicyViolation, match="resolved=0"): + enforce_classification_policy({"resolved": 0}, result) + + +def test_unknown_cannot_be_full_sft() -> None: + result = _classification() + result["source_outcome_class"] = "UNVERIFIED" + result["qc_decision"] = "HOLD_UNVERIFIED" + result["qc_passed"] = False + with pytest.raises(PolicyViolation, match="resolved=-1"): + enforce_classification_policy({"resolved": -1}, result) + + +def _repair_result() -> dict: + """Return a minimal non-mutating repair result for policy tests.""" + + return { + "repair_decision": "NO_CHANGE", + "maximum_training_use": "HOLD", + "invariants": { + "resolved_unchanged": True, + "tool_outputs_unchanged": True, + "model_patch_unchanged": True, + "reference_patch_unchanged": True, + "no_synthetic_execution_result": True, + }, + "operations": [], + "requires_second_review": True, + } + + +def test_nonmutating_repair_decision_rejects_operations() -> None: + """A hold decision cannot smuggle in an operation the applier will ignore.""" + + result = _repair_result() + result["repair_decision"] = "REQUIRES_EXECUTION" + result["operations"] = [ + { + "op": "REWRITE_FINAL_SUMMARY", + "target_turns": [3], + "preconditions": ["Turn 3 is an assistant summary"], + "replacement": "Honest summary", + "reason": "Align the claim with evidence", + } + ] + with pytest.raises(PolicyViolation, match="must not contain"): + enforce_repair_policy(result) + + +def test_step_example_requires_operations() -> None: + """Step-only salvage must include a concrete correction and truncation.""" + + result = _repair_result() + result["repair_decision"] = "CREATE_STEP_EXAMPLE" + result["maximum_training_use"] = "SFT_STEP_ONLY" + with pytest.raises(PolicyViolation, match="requires at least one"): + enforce_repair_policy(result) diff --git a/tests/test_repair.py b/tests/test_repair.py new file mode 100644 index 0000000..19568a4 --- /dev/null +++ b/tests/test_repair.py @@ -0,0 +1,54 @@ +"""Tests for deterministic static repair application and evidence preservation.""" + +from __future__ import annotations + +from swe_data_processing.repair import apply_static_repair + + +def _plan() -> dict: + return { + "sample_id": "sample-1", + "repair_decision": "APPLY_STATIC_REPAIR", + "maximum_training_use": "ERROR_ANALYSIS", + "invariants": { + "resolved_unchanged": True, + "tool_outputs_unchanged": True, + "model_patch_unchanged": True, + "reference_patch_unchanged": True, + "no_synthetic_execution_result": True, + }, + "operations": [ + { + "op": "REWRITE_FINAL_SUMMARY", + "target_turns": [5], + "preconditions": ["Turn 5 is an assistant-only summary."], + "replacement": "The build passed, but functional behavior remains unverified.", + "reason": "Remove an unsupported success claim.", + } + ], + "requires_second_review": True, + "summary": "Correct the final claim without changing execution evidence.", + } + + +def test_rewrite_summary_preserves_tool_output_and_patch() -> None: + record = { + "trajectory_id": "sample-1", + "resolved": -1, + "trajectory": [ + {"role": "system", "content": "system"}, + {"role": "user", "content": "issue"}, + {"role": "assistant", "content": "build", "tool_calls": []}, + {"role": "tool", "content": "build completed with exit code 0"}, + {"role": "assistant", "content": "All tests pass.", "tool_calls": []}, + ], + "metadata": { + "model_patch": {"patch": "+++ b/src/a.py\n"}, + "reference_patch": {"patch": "+++ b/src/a.py\n"}, + }, + } + repaired, diff = apply_static_repair(record, _plan()) + assert repaired["trajectory"][3] == record["trajectory"][3] + assert repaired["trajectory"][4]["content"].startswith("The build passed") + assert repaired["metadata"] == record["metadata"] + assert diff["replaced_original_turn_ids"] == [5]