From 5ab32fd7d70d379b0367a3285f3a10370a2daf50 Mon Sep 17 00:00:00 2001 From: DeepWolf42 <306781384+DeepWolf42@users.noreply.github.com> Date: Sun, 1 Nov 2026 15:17:26 +0100 Subject: [PATCH] Validate Codex completion and provide actionable handoff diagnostics --- docs/CODEX_PROVIDER_OUTPUT.de.md | 79 +++++++++ docs/CODEX_PROVIDER_OUTPUT.md | 76 +++++++++ tests/test_codex_completion_boundary.py | 203 ++++++++++++++++++++++++ tests/test_handoff_cli.py | 6 +- tests/test_verification_streams.py | 44 +++++ tools/harness/codex_completion.py | 97 +++++++++++ tools/harness/handoff_cli.py | 89 +++++++++-- tools/harness/verification_process.py | 9 +- tools/harness/verification_registry.py | 36 +++-- 9 files changed, 605 insertions(+), 34 deletions(-) create mode 100644 docs/CODEX_PROVIDER_OUTPUT.de.md create mode 100644 docs/CODEX_PROVIDER_OUTPUT.md create mode 100644 tests/test_codex_completion_boundary.py create mode 100644 tests/test_verification_streams.py create mode 100644 tools/harness/codex_completion.py diff --git a/docs/CODEX_PROVIDER_OUTPUT.de.md b/docs/CODEX_PROVIDER_OUTPUT.de.md new file mode 100644 index 0000000..5afb35b --- /dev/null +++ b/docs/CODEX_PROVIDER_OUTPUT.de.md @@ -0,0 +1,79 @@ +[English](CODEX_PROVIDER_OUTPUT.md) · **Deutsch** + +# Codex-Vorschläge: vollständiger Abschluss und verständliche Fehler + +Eine vorhandene Vorschlagsdatei reicht nicht zur Übernahme eines ChangeSets. +Der lokale Codex-Adapter verlangt einen vollständigen, widerspruchsfreien +CLI-Ereignisstrom. Die bestehenden Prüfungen für Tests, Rechte, Auftragsumfang, +Dispatch und Verifikationsbelege bleiben zusätzlich erforderlich. + +## Wann ein Vorschlag angenommen wird + +- Genau ein Thread und ein gestarteter Turn, danach eine abgeschlossene finale + Agentennachricht und genau ein erfolgreiches `turn.completed`. +- Nur Reasoning- und Agentennachrichten. Werkzeuge, Fehlermeldungen, fehlgeschlagene + Turns, unbekannte Ereignisse, doppelte/unfertige Elemente und nachträgliche + Aktivität werden zurückgewiesen. +- Finale Nachricht und Ausgabedatei müssen denselben geparsten JSON-Vorschlag + enthalten. Formatierung und Schlüsselreihenfolge dürfen abweichen, Typen und + Werte nicht. Die vorhandene ChangeSet-Prüfung kontrolliert weiterhin Schema, + Pfade, Hashes und angeforderte Tests. +- Fehlende Verbrauchsdaten bleiben unbekannt. Vorhandene Tokenzahlen müssen + nichtnegative ganze Zahlen sein; Wahrheitswerte sind keine gültigen Tokenzahlen. + Fehlender Verbrauch oder Geldkosten werden nicht als null ausgegeben. + +Der vorhandene Prozessrunner behält seine Frist, Ausgabegrenze und Begrenzung +der Prozessnachkommen. Die Vorschlagsdatei wird mit höchstens2MiB+1 Bytes gelesen; +der angenommene Ereignisstrom ist auf1MiB und4096 nichtleere Zeilen begrenzt. +Keine neue Prozessverwaltung, Abhängigkeit, Zugangsdaten oder Hintergrunddienste. + +## Fehler bleiben nachvollziehbar + +Ungültige oder zu große Vorschläge und widersprüchliche Ereignisse ergeben einen +fehlgeschlagenen WorkerResult mit festem `codex:CLI_*`-Code. Der vorhandene Handoff +speichert ihn vor einer Änderung als `FAILED_WORKER`. Dieselbe Auftragskennung +ruft anschließend das gespeicherte Ergebnis ab, statt die KI erneut zu starten. +Unterbrechungen bleiben wie bisher UNKNOWN ohne automatische Wiederholung. +Rohe CLI-Diagnosen werden nicht in Workerbegründung oder Textanzeige übernommen. + +Beispiele: `codex:CLI_INCOMPLETE_TURN`, `codex:CLI_REPORTED_FAILURE`, +`codex:CLI_PROPOSAL_MISMATCH`, `codex:CLI_INVALID_PROPOSAL`. Ursache und gespeicherten +Zustand prüfen, bevor ein gesondert freigegebener neuer Versuch erwogen wird. +Die Diagnose erteilt keine neue Wiederholungs-, Commit-, Push- oder Mergefreigabe. + +## Bedienung + +Standard bleibt JSON. Statuswerte und Exit-Codes bleiben kompatibel:0 bei +`DONE_VERIFIED`,3 bei erfolglosem gespeicherten Ergebnis,2 bei Eingabe-/Laufzeitfehler. +Vorprüfungsfehler behalten ihre Ausnahme-Klasse in `reason`; hinzu kommt ein fester +`diagnostic_code`, etwa `STALE_SOURCE_REVISION`, `SOURCE_NOT_CLEAN`, +`COMMIT_NOT_AUTHORIZED`, `TASK_ID_CONFLICT` oder `INVALID_INPUT_SHAPE`. + +Für eine verständliche deutsche Ausgabe gibt es bei beiden Operationen `--human`: + +```text +python -B -m tools.harness.handoff_cli status --input request.json --profile project.local.json --human +``` + +Angezeigt werden feste Erklärungen und gegebenenfalls ein syntaktisch gültiger +Kandidatenhash, keine beliebige Modellbegründung. Programme verwenden weiter JSON. + +## Nachweise und Grenzen + +Gegenfälle prüfen unvollständige/fehlgeschlagene/mehrdeutige Streams, abweichende +Vorschläge, ungültige JSON-/Verbrauchsdaten, unerlaubte Werkzeuge, gespeicherte +Fehlversuche und CLI-Kompatibilität. Deterministische Prozessfixtures bleiben +Test-Doubles. Ein echter CLI-Kompatibilitätslauf ist ein gesonderter Nachweis. + +Der vorhandene Runner bietet jetzt optional `separate_stderr=True`, hier nur für +diesen Adapter. Standardverhalten mit gemischter Ausgabe und Ergebnisschema +bleiben unverändert. Beide Kanäle teilen eine Ausgabegrenze und Prozessbegrenzung; +stderr wird begrenzt gelesen, aber nicht als Vorschlagsereignis oder Begründung +zurückgegeben. Ungültige UTF-8-Zeichen werden weiterhin ersetzt: kein Nachweis des +bytegenauen stdout-Kanals. Nicht-JSON-Daten in stdout führen zur Ablehnung; +unbekannte CLI-/Ereignisformate brauchen eine ausdrückliche neue Prüfung. +Ein gültiger Stream ist keine Quellattestation, +Datenschutz-Sandbox, signierter Beleg oder allgemeiner Modellqualitätsnachweis. + +Offizielle Schnittstellenreferenz, geprüft01.11.2026: +[Codex non-interaktive JSON-Ausgabe](https://learn.chatgpt.com/docs/non-interactive-mode). diff --git a/docs/CODEX_PROVIDER_OUTPUT.md b/docs/CODEX_PROVIDER_OUTPUT.md new file mode 100644 index 0000000..f2dfc14 --- /dev/null +++ b/docs/CODEX_PROVIDER_OUTPUT.md @@ -0,0 +1,76 @@ +**English** · [Deutsch](CODEX_PROVIDER_OUTPUT.de.md) + +# Codex proposal completion and handoff diagnostics + +A proposal file alone must not authorize a ChangeSet. The local Codex adapter +accepts a proposal only after one complete, consistent CLI JSON event stream. +This is an adapter acceptance condition; registered verification and the existing +policy, dispatch, scope and receipt checks remain mandatory. + +## Accepted completion + +- One thread and one started turn, followed by one completed final agent message + and one successful `turn.completed`. +- Only reasoning and agent-message items. Tools, errors, failed turns, unknown + events, duplicate/unfinished items and activity after completion are rejected. +- The final message and output file must contain the same parsed JSON proposal. + JSON formatting/key order may differ; JSON types and values must match. +- Existing ChangeSet validation still checks the actual schema, paths, hashes + and requested verification. Stream completion does not make model text true. +- Missing usage remains unknown. Present token counts must be nonnegative + integers, excluding booleans; no zero usage or monetary cost is inferred. + +The existing process runner retains its deadline, output cap and descendant +containment. The adapter reads at most2MiB+1 proposal bytes; the accepted trace +is bounded to1MiB and4096 nonempty lines. No new process runner, dependencies, +daemon, background retry or credentials are introduced. + +## Failures and recovery + +Malformed/oversized proposals and inconsistent event streams return a failed +WorkerResult with a fixed `codex:CLI_*` code. The existing handoff records it as +`FAILED_WORKER` before applying changes. Repeating that task ID retrieves its +stored result, not another provider call. Ambiguous interruptions retain the +existing UNKNOWN/no-replay behavior. Raw CLI diagnostic text is not copied into +the worker rationale or human display. + +Examples: `codex:CLI_INCOMPLETE_TURN`, `codex:CLI_REPORTED_FAILURE`, +`codex:CLI_PROPOSAL_MISMATCH`, `codex:CLI_INVALID_PROPOSAL`. Inspect the local +saved result and reconcile the cause before considering a separately authorized +new attempt. These codes grant no new retry, commit, push or merge authority. + +## Operator output + +JSON remains the default. Existing status values and exit codes remain0 for +`DONE_VERIFIED`,3 for unsuccessful stored results,2 for input/runtime errors. +Preflight errors retain their exception-class `reason` and add a bounded +`diagnostic_code`, including `STALE_SOURCE_REVISION`, `SOURCE_NOT_CLEAN`, +`COMMIT_NOT_AUTHORIZED`, `TASK_ID_CONFLICT` and `INVALID_INPUT_SHAPE`. + +For German human-readable output, add `--human` to either operation: + +```text +python -B -m tools.harness.handoff_cli status --input request.json --profile project.local.json --human +``` + +This format displays fixed explanations and a syntactically valid candidate hash, +never arbitrary model rationale. Machine consumers should keep using default JSON. + +## Verification and limits + +Regression tests cover incomplete/failed/ambiguous streams, proposal mismatch, +invalid JSON/usage, forbidden tool items, durable failed-attempt retrieval and +CLI compatibility. Deterministic process fixtures are labeled test doubles. +A real CLI compatibility trial is separate from these tests. + +The existing runner now offers optional `separate_stderr=True`, used only by this +adapter. Its default combined-output behavior and result schema are unchanged. +Both streams share one output budget and process scope; stderr is bounded and +drained but never interpreted as proposal events or returned in the rationale. +The runner still decodes UTF-8 with replacement, so this establishes no byte-level +stdout attestation. Non-JSON data in stdout fails closed; an unsupported CLI/event +format needs explicit requalification instead of silent acceptance. A valid stream is not a source +attestation, privacy sandbox, signed receipt or general model-quality guarantee. + +Official interface reference, checked2026-11-01: +[Codex non-interactive JSON output](https://learn.chatgpt.com/docs/non-interactive-mode). diff --git a/tests/test_codex_completion_boundary.py b/tests/test_codex_completion_boundary.py new file mode 100644 index 0000000..cdf47e4 --- /dev/null +++ b/tests/test_codex_completion_boundary.py @@ -0,0 +1,203 @@ +"""Provider regressions: a file is not evidence of a completed CLI turn.""" +import copy +import io +import json +from contextlib import redirect_stdout +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch + +from tools.harness.handoff_cli import CodexProvider, main +from tools.harness.verification_registry import VerificationResult +from tools.harness.worker_provider import WorkerRequest, WorkerStatus + + +class CodexCompletionBoundaryTests(unittest.TestCase): + def setUp(self): + self.proposal={'schema':'loop42.changeset.v1','changes':[{'operation':'modify','path':'answer.py', + 'base_sha256':'0'*64,'search':'wrong','replace':'right','expected_matches':1}], + 'requested_verification':['test:answer']} + self.events=[{'type':'thread.started','thread_id':'synthetic-thread'}, + {'type':'turn.started'}, + {'type':'item.completed','item':{'id':'final','type':'agent_message','text':json.dumps(self.proposal)}}, + {'type':'turn.completed','usage':{'input_tokens':12,'output_tokens':3}}] + self.request=WorkerRequest('example','attempt-1','a'*40,'b'*64,'Fix answer', + '{"files":{"answer.py":"wrong"},"file_sha256":{"answer.py":"'+ '0'*64 +'"},"verification_ids":["test:answer"]}', + ('propose_changes','request_verification'),'coding') + + def call(self,events=None,raw=None,file_data=None,exit_code=0): + data=json.dumps(self.proposal).encode() if file_data is None else file_data + stream='\n'.join(json.dumps(e) for e in (self.events if events is None else events)) if raw is None else raw + def process(workspace,identifiers,registry,**kwargs): + argv=registry.specs[0].argv + Path(argv[argv.index('--output-last-message')+1]).write_bytes(data) + return (VerificationResult('provider:codex',argv,exit_code,False,stream,False),) + with patch('tools.harness.handoff_cli.shutil.which',return_value='codex'),patch('tools.harness.handoff_cli.run_verifications',side_effect=process): + return CodexProvider({'allow_codex':True})(self.request) + + def rejected(self,**kwargs): + result=self.call(**kwargs) + self.assertEqual(result.status,WorkerStatus.FAILED) + self.assertIsNone(result.changeset) + return result + + def test_complete_bound_turn_preserves_usage(self): + result=self.call() + self.assertEqual(result.status,WorkerStatus.PROPOSAL) + self.assertEqual((result.usage.input_tokens,result.usage.output_tokens),(12,3)) + + def test_missing_completion_failed_turn_and_explicit_error_reject_existing_file(self): + cases=[self.events[:-1],self.events[:-1]+[{'type':'turn.failed','error':{'message':'synthetic failure'}}], + self.events[:-1]+[{'type':'error','message':'synthetic failure'},self.events[-1]]] + for events in cases: + with self.subTest(events=events):self.rejected(events=events) + + def test_ambiguous_order_duplicate_completion_and_post_completion_events_rejected(self): + cases=[self.events+[self.events[-1]],self.events[:2]+[self.events[-1],self.events[2]], + self.events+[{'type':'item.completed','item':{'id':'late','type':'reasoning','text':'late'}}], + self.events[:2]+[self.events[2],self.events[2],self.events[-1]]] + for events in cases: + with self.subTest(events=events):self.rejected(events=events) + + def test_unknown_or_non_object_events_and_duplicate_json_keys_rejected(self): + for event in ({'type':'future.unknown'},[],{'type':'item.completed','item':None}): + with self.subTest(event=event):self.rejected(events=self.events[:2]+[event]+self.events[2:]) + raw='\n'.join(json.dumps(e) for e in self.events[:-1])+'\n{"type":"turn.failed","type":"turn.completed","usage":{}}' + self.rejected(raw=raw) + + def test_non_json_stdout_is_not_ignored_even_with_a_valid_completion(self): + self.rejected(raw='unstructured stdout\n'+'\n'.join(json.dumps(e) for e in self.events)) + + def test_message_and_file_must_bind_to_the_same_proposal(self): + alternate=copy.deepcopy(self.proposal);alternate['changes'][0]['replace']='different' + self.rejected(file_data=json.dumps(alternate).encode()) + + def test_invalid_or_oversized_output_becomes_failure_not_exception(self): + for data in (b'{invalid',b'[]',b'x'*(2*1024*1024+1)): + with self.subTest(length=len(data)):self.rejected(file_data=data) + + def test_invalid_usage_does_not_escape_as_exception_or_authority(self): + for value in (True,-1,'12'): + events=copy.deepcopy(self.events);events[-1]['usage']['input_tokens']=value + with self.subTest(value=value):self.rejected(events=events) + + def test_missing_usage_is_unknown_not_invented_zero(self): + events=copy.deepcopy(self.events);events[-1]['usage']={} + result=self.call(events=events) + self.assertEqual(result.status,WorkerStatus.PROPOSAL) + self.assertIsNone(result.usage.input_tokens) + + def test_unfinished_item_or_tools_never_authorize_a_proposal(self): + events=self.events[:2]+[{'type':'item.started','item':{'id':'pending','type':'reasoning'}}]+self.events[2:] + self.rejected(events=events) + events=self.events[:2]+[{'type':'item.completed','item':{'id':'tool','type':'command_execution'}}]+self.events[2:] + self.rejected(events=events) + + def test_raw_provider_diagnostics_are_not_returned_to_callers(self): + result=self.rejected(raw='SYNTHETIC_PRIVATE_DIAGNOSTIC',exit_code=1) + self.assertNotIn('SYNTHETIC_PRIVATE_DIAGNOSTIC',result.rationale_summary) + + def test_final_json_binding_is_semantic_but_preserves_json_types(self): + result=self.call(file_data=json.dumps(self.proposal,indent=2,sort_keys=True).encode()) + self.assertEqual(result.status,WorkerStatus.PROPOSAL) + events=copy.deepcopy(self.events) + different=copy.deepcopy(self.proposal);different['changes'][0]['expected_matches']=True + events[2]['item']['text']=json.dumps(different) + self.rejected(events=events) + + def test_valid_reasoning_lifecycle_does_not_grant_extra_authority(self): + events=self.events[:2]+[ + {'type':'item.started','item':{'id':'reason','type':'reasoning'}}, + {'type':'item.updated','item':{'id':'reason','type':'reasoning','text':'Untrusted reasoning text'}}, + {'type':'item.completed','item':{'id':'reason','type':'reasoning','text':'Not evidence'}}]+self.events[2:] + self.assertEqual(self.call(events=events).status,WorkerStatus.PROPOSAL) + + def test_provider_failure_is_persisted_and_duplicate_does_not_relaunch(self): + import hashlib + import test_handoff as fixtures + from tools.harness.handoff import execute,status,encoded + fixture=fixtures.HandoffTests('test_real_git_red_green_commit_and_separate_status') + fixture.setUp();self.addCleanup(fixture.doCleanups) + calls=[] + def process(workspace,identifiers,registry,**kwargs): + calls.append(1);argv=registry.specs[0].argv + proposal=copy.deepcopy(self.proposal) + proposal['changes'][0].update(path='answer.txt',base_sha256=hashlib.sha256(b'wrong\n').hexdigest()) + Path(argv[argv.index('--output-last-message')+1]).write_text(json.dumps(proposal)) + return (VerificationResult('provider:codex',argv,0,False,'{"type":"turn.failed"}',False),) + with patch('tools.harness.handoff_cli.shutil.which',return_value='codex'),patch('tools.harness.handoff_cli.run_verifications',side_effect=process): + provider=CodexProvider({'allow_codex':True}) + result=execute(fixture.request,fixture.profile,provider=provider) + self.assertEqual(result['status'],'FAILED_WORKER') + saved=json.loads(encoded(result)) + self.assertEqual(status(fixture.request,fixture.profile),saved) + self.assertEqual(execute(fixture.request,fixture.profile,provider=provider),saved) + self.assertEqual(len(calls),1) + self.assertNotIn('commit',result) + self.assertNotIn('green',result) + self.assertEqual((fixture.repo/'answer.txt').read_bytes(),b'wrong\n') + self.assertEqual((Path(result['workspace'])/'answer.txt').read_bytes(),b'wrong\n') + + +class HandoffHumanDiagnosticsTests(unittest.TestCase): + def invoke(self,request,profile,extra=(),result=None,error=None): + with tempfile.TemporaryDirectory() as directory: + folder=Path(directory) + for name,value in (('input.json',request),('profile.json',profile)): + (folder/name).write_text(json.dumps(value),encoding='utf-8') + output=io.StringIO() + with patch('tools.harness.handoff_cli.execute',return_value=result,side_effect=error),redirect_stdout(output): + code=main(['run','--input',str(folder/'input.json'),'--profile',str(folder/'profile.json'),*extra]) + return code,output.getvalue() + + def test_invalid_input_shape_is_json_failure_without_traceback(self): + for request,profile in (({},[]),([],{})): + with self.subTest(request=request,profile=profile): + code,output=self.invoke(request,profile) + value=json.loads(output) + self.assertEqual(code,2) + self.assertEqual(value['status'],'UNKNOWN') + self.assertEqual(value['diagnostic_code'],'INVALID_INPUT_SHAPE') + + def test_stale_source_has_actionable_code_without_private_error_text(self): + code,output=self.invoke({}, {},error=ValueError('stale source revision')) + value=json.loads(output) + self.assertEqual(code,2) + self.assertEqual(value['reason'],'ValueError') + self.assertEqual(value['diagnostic_code'],'STALE_SOURCE_REVISION') + + def test_human_output_is_german_and_does_not_render_worker_text(self): + code,output=self.invoke({}, {},extra=['--human'],result={'status':'FAILED_WORKER', + 'worker':{'rationale_summary':'\u001b[31mSYNTHETIC_PRIVATE_DIAGNOSTIC'}}) + self.assertEqual(code,3) + self.assertIn('Worker',output) + self.assertNotIn('SYNTHETIC_PRIVATE_DIAGNOSTIC',output) + self.assertNotIn('\u001b',output) + + def test_default_json_result_and_exit_code_remain_compatible(self): + expected={'status':'DONE_VERIFIED','commit':'a'*40} + code,output=self.invoke({}, {},result=expected) + self.assertEqual(code,0) + self.assertEqual(json.loads(output),expected) + + def test_new_human_mode_preserves_failed_attempt_exit_code(self): + code,output=self.invoke({}, {},extra=['--human'],result={'status':'UNKNOWN','reason':'interrupted_attempt_no_replay'}) + self.assertEqual(code,3) + self.assertIn('Unbekannt',output) + + def test_real_cli_process_invalid_profile_has_no_traceback(self): + import subprocess,sys,os + with tempfile.TemporaryDirectory() as directory: + folder=Path(directory) + (folder/'input.json').write_text('{}') + (folder/'profile.json').write_text('[]') + env=dict(os.environ,PYTHONDONTWRITEBYTECODE='1') + process=subprocess.run([sys.executable,'-B','-m','tools.harness.handoff_cli','status', + '--input',str(folder/'input.json'),'--profile',str(folder/'profile.json')], + cwd=Path(__file__).resolve().parents[1],env=env,capture_output=True,timeout=10) + self.assertEqual(process.returncode,2) + self.assertEqual(json.loads(process.stdout)['diagnostic_code'],'INVALID_INPUT_SHAPE') + self.assertNotIn(b'Traceback',process.stderr) + +if __name__=='__main__':unittest.main() diff --git a/tests/test_handoff_cli.py b/tests/test_handoff_cli.py index 1969343..7d6db46 100644 --- a/tests/test_handoff_cli.py +++ b/tests/test_handoff_cli.py @@ -33,13 +33,15 @@ def test_cli_proposal_is_parsed_and_tool_events_are_rejected(self): 'requested_verification':['test:answer']} for tool in (False,True): def transport(workspace, identifiers, registry, **kw): + self.assertIs(kw.get('separate_stderr'),True) argv=registry.specs[0].argv self.assertEqual(argv[argv.index('--sandbox')+1],'read-only') schema=json.loads(Path(argv[argv.index('--output-schema')+1]).read_bytes()) self.assertIn('anyOf',schema['properties']['changes']['items']) Path(argv[argv.index('--output-last-message')+1]).write_text(json.dumps(proposal)) - event={'type':'item.completed','item':{'type':'command_execution' if tool else 'agent_message'}} - return (VerificationResult('provider:codex',argv,0,False,json.dumps(event),False),) + event={'type':'item.completed','item':{'id':'final','type':'command_execution' if tool else 'agent_message','text':json.dumps(proposal)}} + events=[{'type':'thread.started','thread_id':'synthetic-thread'},{'type':'turn.started'},event,{'type':'turn.completed','usage':{}}] + return (VerificationResult('provider:codex',argv,0,False,'\n'.join(json.dumps(row) for row in events),False),) with patch('tools.harness.handoff_cli.shutil.which',return_value='codex'), patch('tools.harness.handoff_cli.run_verifications',side_effect=transport): result=CodexProvider({'allow_codex':True})(self.request()) self.assertEqual(result.status,WorkerStatus.FAILED if tool else WorkerStatus.PROPOSAL) diff --git a/tests/test_verification_streams.py b/tests/test_verification_streams.py new file mode 100644 index 0000000..d454feb --- /dev/null +++ b/tests/test_verification_streams.py @@ -0,0 +1,44 @@ +"""Real subprocess checks for optional stdout/stderr separation, not model mocks.""" +from pathlib import Path +import sys +import tempfile +import unittest + +from tools.harness.verification_registry import VerificationRegistry,VerificationSpec,run_verifications + + +class VerificationStreamsTests(unittest.TestCase): + def run_process(self,script,**kwargs): + registry=VerificationRegistry((VerificationSpec('test:streams',(sys.executable,'-B','-c',script),1),)) + with tempfile.TemporaryDirectory() as directory: + return run_verifications(Path(directory),('test:streams',),registry,**kwargs)[0] + + def test_separated_stderr_cannot_corrupt_json_stdout(self): + result=self.run_process('import sys;print("{\\"ok\\":true}");print("private diagnostic",file=sys.stderr)',separate_stderr=True) + self.assertTrue(result.passed) + self.assertEqual(result.output.strip(),'{"ok":true}') + self.assertNotIn('private diagnostic',result.output) + + def test_default_combined_output_remains_compatible(self): + result=self.run_process('import sys;print("stdout");print("stderr",file=sys.stderr)') + self.assertTrue(result.passed) + self.assertIn('stdout',result.output) + self.assertIn('stderr',result.output) + + def test_stderr_counts_toward_the_shared_limit(self): + result=self.run_process('import sys;print("{\\"ok\\":true}");sys.stderr.write("x"*4096)',separate_stderr=True,output_limit=64) + self.assertFalse(result.passed) + self.assertTrue(result.output_truncated) + self.assertIsNone(result.exit_code) + self.assertLessEqual(len(result.output.encode()),64) + + def test_timeout_still_fails_closed_with_two_streams(self): + result=self.run_process('import time;time.sleep(30)',separate_stderr=True) + self.assertFalse(result.passed) + self.assertTrue(result.timed_out) + self.assertIsNone(result.exit_code) + + def test_invalid_option_is_rejected_before_execution(self): + with self.assertRaises(TypeError):self.run_process('raise SystemExit(0)',separate_stderr=1) + +if __name__=='__main__':unittest.main() diff --git a/tools/harness/codex_completion.py b/tools/harness/codex_completion.py new file mode 100644 index 0000000..b6eca65 --- /dev/null +++ b/tools/harness/codex_completion.py @@ -0,0 +1,97 @@ +"""Validate one bounded Codex JSON event stream before accepting its proposal.""" +import json +from tools.harness.ollama_worker import strict_json + +MAX_TRACE_BYTES=1024*1024 +MAX_PROPOSAL_BYTES=2*1024*1024 + + +class CodexCompletionError(ValueError): + """Only fixed diagnostic codes, never provider text, cross this boundary.""" + + +def reject(code): + raise CodexCompletionError(code) + + +def canonical(value): + return json.dumps(value,sort_keys=True,separators=(',',':'),ensure_ascii=True,allow_nan=False) + + +def completed_usage(trace, proposal): + if not isinstance(trace,str) or len(trace.encode('utf-8'))>MAX_TRACE_BYTES: + reject('CLI_TRACE_BUDGET') + lines=[line for line in trace.splitlines() if line.strip()] + if len(lines)>4096: + reject('CLI_TRACE_BUDGET') + thread=False;started=False;complete=False;final=False + pending={};closed=set();usage={} + for line in lines: + if complete: + reject('CLI_EVENT_AFTER_COMPLETION') + try: + event=strict_json(line.encode('utf-8')) + except (ValueError,TypeError,RecursionError): + reject('CLI_INVALID_EVENT') + if not isinstance(event,dict) or not isinstance(event.get('type'),str): + reject('CLI_INVALID_EVENT') + kind=event['type'] + if kind in ('error','turn.failed'): + reject('CLI_REPORTED_FAILURE') + if kind=='thread.started': + if thread or started or not isinstance(event.get('thread_id'),str) or not 1<=len(event['thread_id'])<=256: + reject('CLI_INVALID_LIFECYCLE') + thread=True + elif kind=='turn.started': + if not thread or started: + reject('CLI_INVALID_LIFECYCLE') + started=True + elif kind in ('item.started','item.updated','item.completed'): + if not started or final: + reject('CLI_INVALID_LIFECYCLE') + item=event.get('item') + if not isinstance(item,dict) or not isinstance(item.get('id'),str) or not 1<=len(item['id'])<=256: + reject('CLI_INVALID_ITEM') + identity=item['id'];item_type=item.get('type') + if item_type not in ('reasoning','agent_message'): + reject('CLI_UNSUPPORTED_ITEM') + if identity in closed: + reject('CLI_DUPLICATE_ITEM') + if kind=='item.started': + if identity in pending: + reject('CLI_DUPLICATE_ITEM') + pending[identity]=item_type + elif kind=='item.updated': + if pending.get(identity)!=item_type: + reject('CLI_INVALID_ITEM') + else: + if identity in pending and pending.pop(identity)!=item_type: + reject('CLI_INVALID_ITEM') + closed.add(identity) + if item_type=='agent_message': + text=item.get('text') + if not isinstance(text,str) or len(text.encode('utf-8'))>MAX_PROPOSAL_BYTES: + reject('CLI_INVALID_FINAL_MESSAGE') + try: + value=strict_json(text.encode('utf-8')) + except (ValueError,TypeError,RecursionError): + reject('CLI_INVALID_FINAL_MESSAGE') + if canonical(value)!=canonical(proposal): + reject('CLI_PROPOSAL_MISMATCH') + final=True + elif kind=='turn.completed': + if not started or not final or pending: + reject('CLI_INVALID_LIFECYCLE') + usage=event.get('usage',{}) + if not isinstance(usage,dict): + reject('CLI_INVALID_USAGE') + for key in ('input_tokens','output_tokens'): + value=usage.get(key) + if value is not None and (type(value) is not int or value<0): + reject('CLI_INVALID_USAGE') + complete=True + else: + reject('CLI_UNSUPPORTED_EVENT') + if not complete: + reject('CLI_INCOMPLETE_TURN') + return {key:usage.get(key) for key in ('input_tokens','output_tokens')} diff --git a/tools/harness/handoff_cli.py b/tools/harness/handoff_cli.py index 40276ef..d5ba942 100644 --- a/tools/harness/handoff_cli.py +++ b/tools/harness/handoff_cli.py @@ -6,6 +6,7 @@ import shutil import sys import tempfile +import re from tools.harness.handoff import execute, status, encoded from tools.harness.change_set import CHANGESET_SCHEMA, parse_changeset @@ -13,6 +14,7 @@ from tools.harness.worker_provider import WorkerResult, WorkerStatus, WorkerUsage from tools.harness.verification_registry import VerificationRegistry, VerificationSpec, run_verifications from tools.harness.worker_dispatch_cli import _read_input +from tools.harness.codex_completion import completed_usage, CodexCompletionError, MAX_PROPOSAL_BYTES class CodexProvider: @@ -29,6 +31,8 @@ def __call__(self, request): common=dict(provider='codex-cli',model='configured-cli-model-unknown',task_id=request.task_id, attempt_id=request.attempt_id,source_revision=request.source_revision, context_fingerprint=request.context_fingerprint) + def failed(code): + return WorkerResult(status=WorkerStatus.FAILED,rationale_summary='codex:'+code,**common) executable=shutil.which('codex') if not executable or self.profile.get('allow_codex') is not True: return WorkerResult(status=WorkerStatus.BLOCKED,rationale_summary='CLI missing or provider use not authorized',**common) @@ -55,43 +59,94 @@ def __call__(self, request): argv=argv[:2]+tuple(x for name in disabled for x in ('--disable',name))+( '-c','web_search="disabled"','-c','project_doc_max_bytes=0')+argv[2:] registry=VerificationRegistry((VerificationSpec('provider:codex',argv,120),)) - run=run_verifications(folder,('provider:codex',),registry,output_limit=1024*1024)[0] + try: + run=run_verifications(folder,('provider:codex',),registry,output_limit=1024*1024,separate_stderr=True)[0] + except (ValueError,OSError,RuntimeError): + return failed('CLI_PROCESS_FAILURE') if not run.passed or not output.is_file() or output.is_symlink(): - return WorkerResult(status=WorkerStatus.FAILED,rationale_summary='CLI failure '+str(run.exit_code)+': '+run.output[-1800:],**common) - data=output.read_bytes() - if len(data)>2*1024*1024: - raise ValueError('proposal exceeds limit') - changes=parse_changeset(strict_json(data)) - usage={} - for line in run.output.splitlines(): - try: - event=json.loads(line) - if event.get('type','').startswith('item.') and event.get('item',{}).get('type') not in ('reasoning','agent_message'): - return WorkerResult(status=WorkerStatus.FAILED,rationale_summary='Unexpected tool or unknown item event',**common) - if event.get('type')=='turn.completed': usage=event.get('usage',{}) - except (ValueError,AttributeError): - pass + return failed('CLI_PROCESS_OR_OUTPUT_FAILED') + try: + with output.open('rb') as stream: + data=stream.read(MAX_PROPOSAL_BYTES+1) + if len(data)>MAX_PROPOSAL_BYTES: + return failed('CLI_PROPOSAL_BUDGET') + proposal=strict_json(data) + usage=completed_usage(run.output,proposal) + changes=parse_changeset(proposal) + except CodexCompletionError as error: + return failed(str(error)) + except (ValueError,TypeError,KeyError,OSError,RecursionError): + return failed('CLI_INVALID_PROPOSAL') return WorkerResult(status=WorkerStatus.PROPOSAL,changeset=changes, rationale_summary='Unverified official CLI proposal; no model authority', raw_response_fingerprint=hashlib.sha256(data).hexdigest(), usage=WorkerUsage(input_tokens=usage.get('input_tokens'),output_tokens=usage.get('output_tokens')),**common) +def diagnostic(error): + messages={'stale source revision':'STALE_SOURCE_REVISION', + 'unclean/untracked source checkout':'SOURCE_NOT_CLEAN', + 'explicit local commit authority required':'COMMIT_NOT_AUTHORIZED', + 'task ID already bound to another request/profile':'TASK_ID_CONFLICT', + 'INVALID_INPUT_SHAPE':'INVALID_INPUT_SHAPE'} + fallback='LOCAL_INPUT_UNAVAILABLE' if isinstance(error,OSError) else 'HANDOFF_RUNTIME_FAILED' if isinstance(error,RuntimeError) else 'HANDOFF_INPUT_INVALID' + return messages.get(str(error),fallback) + + +def human_result(result): + texts={'DONE_VERIFIED':'Geprüft: Der registrierte Test bestätigt den gespeicherten Kandidaten.', + 'FAILED_WORKER':'Worker fehlgeschlagen. Vor einem neuen Versuch Ursache und gespeicherten Zustand prüfen.', + 'FAILED_VERIFICATION':'Prüfung fehlgeschlagen. Der Kandidat bleibt zur Untersuchung erhalten.', + 'BLOCKED_CAPABILITY':'Blockiert: Kein verfügbarer und freigegebener Worker.', + 'BLOCKED_SCOPE':'Blockiert: Der Vorschlag überschreitet den erlaubten Umfang.', + 'RED_NOT_REPRODUCED':'Der Ausgangsfehler wurde nicht reproduziert. Keine Änderung ausgeführt.', + 'UNKNOWN':'Unbekannt: Es liegt kein aktueller belastbarer Erfolgsnachweis vor.'} + lines=[texts.get(result.get('status'),'Unbekannter Status. Gespeicherte Belege prüfen.')] + explanations={'STALE_SOURCE_REVISION':'Die Quelle hat einen anderen Stand. Auftrag gegen die aktuelle Revision neu abgleichen.', + 'SOURCE_NOT_CLEAN':'Die Quelle enthält Änderungen oder untracked Dateien. Vorhandene Arbeit erhalten und zuerst abgleichen.', + 'COMMIT_NOT_AUTHORIZED':'Die lokale Commit-Freigabe fehlt im vertrauenswürdigen Profil.', + 'TASK_ID_CONFLICT':'Diese Auftragskennung ist bereits an andere Eingaben gebunden.', + 'INVALID_INPUT_SHAPE':'Auftrag und Profil müssen JSON-Objekte sein.'} + if result.get('diagnostic_code') in explanations:lines.append(explanations[result['diagnostic_code']]) + recovery_texts={'interrupted_attempt_no_replay':'Der gespeicherte Versuch ist unterbrochen. Zustand prüfen; nicht automatisch erneut starten.', + 'incomplete_claim_no_replay':'Die gespeicherte Auftragsbindung ist unvollständig. Vor einem neuen Versuch abgleichen.', + 'no_claim_or_receipt':'Für diesen Auftrag ist noch kein gespeicherter Versuch vorhanden.', + 'saved_receipt_not_current_or_complete':'Der gespeicherte Beleg ist nicht vollständig oder nicht mehr aktuell. Quellen und Prüfer neu abgleichen.'} + reason=result.get('reason') + if isinstance(reason,str) and reason in recovery_texts:lines.append(recovery_texts[reason]) + worker=result.get('worker') + if isinstance(worker,dict): + provider_texts={'codex:CLI_INCOMPLETE_TURN':'Der nachgewiesene Workerabschluss fehlt. Ein vorhandener Vorschlag reicht nicht.', + 'codex:CLI_REPORTED_FAILURE':'Die CLI meldete einen Fehler. Der Vorschlag wurde verworfen.', + 'codex:CLI_PROPOSAL_MISMATCH':'Abschlussmeldung und Vorschlagsdatei widersprechen sich.', + 'codex:CLI_INVALID_PROPOSAL':'Der Vorschlag erfüllt das erforderliche Format nicht.', + 'codex:CLI_PROCESS_OR_OUTPUT_FAILED':'CLI-Prozess oder begrenzte Ausgabedatei ist fehlgeschlagen.'} + rationale=worker.get('rationale_summary') + if isinstance(rationale,str) and rationale in provider_texts:lines.append(provider_texts[rationale]) + commit=result.get('commit') + if isinstance(commit,str) and re.fullmatch('[0-9a-f]{40}',commit):lines.append('Kandidat: '+commit) + return '\n'.join(lines) + + def main(argv=None): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('operation',choices=('run','status')) parser.add_argument('--input',required=True) parser.add_argument('--profile',required=True,help='Trusted operator-owned local JSON; never use an issue as this profile') + parser.add_argument('--human',action='store_true',help='German status text; default remains machine-readable JSON') args=parser.parse_args(argv) try: request=strict_json(_read_input(args.input)) profile=strict_json(_read_input(args.profile)) + if not isinstance(request,dict) or not isinstance(profile,dict): + raise TypeError('INVALID_INPUT_SHAPE') provider=CodexProvider(profile) if profile.get('allow_codex') is True and shutil.which('codex') else None result=status(request,profile) if args.operation=='status' else execute(request,profile,provider=provider) - print(json.dumps(result,ensure_ascii=True,allow_nan=False)) + print(human_result(result) if args.human else json.dumps(result,ensure_ascii=True,allow_nan=False)) return 0 if result['status']=='DONE_VERIFIED' else 3 except (ValueError,TypeError,KeyError,OSError,RuntimeError) as error: - print(json.dumps({'status':'UNKNOWN','reason':type(error).__name__})) + result={'status':'UNKNOWN','reason':type(error).__name__,'diagnostic_code':diagnostic(error)} + print(human_result(result) if args.human else json.dumps(result)) return 2 diff --git a/tools/harness/verification_process.py b/tools/harness/verification_process.py index dc31d14..59219d3 100644 --- a/tools/harness/verification_process.py +++ b/tools/harness/verification_process.py @@ -45,7 +45,9 @@ def _checked(value): class VerificationProcess: - def __init__(self, argv: tuple[str, ...], workspace: Path): + def __init__(self, argv: tuple[str, ...], workspace: Path, *, separate_stderr: bool = False): + if type(separate_stderr) is not bool: + raise TypeError('separate_stderr must be boolean') self.job = None self.process = None if os.name == 'nt': @@ -62,7 +64,7 @@ def __init__(self, argv: tuple[str, ...], workspace: Path): self.process = subprocess.Popen( [sys.executable, str(Path(__file__).resolve()), name, json.dumps(argv)], cwd=workspace, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, shell=False, + stderr=subprocess.PIPE if separate_stderr else subprocess.STDOUT, shell=False, ) except BaseException: self.api.CloseHandle(self.job) @@ -70,7 +72,8 @@ def __init__(self, argv: tuple[str, ...], workspace: Path): raise else: self.process = subprocess.Popen(argv, cwd=workspace, stdin=subprocess.DEVNULL, - stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=False, start_new_session=True) + stdout=subprocess.PIPE, stderr=subprocess.PIPE if separate_stderr else subprocess.STDOUT, + shell=False, start_new_session=True) def terminate(self): if self.job is not None: diff --git a/tools/harness/verification_registry.py b/tools/harness/verification_registry.py index 8e82835..54bf1ca 100644 --- a/tools/harness/verification_registry.py +++ b/tools/harness/verification_registry.py @@ -96,12 +96,15 @@ def run_verifications( *, stop_on_failure: bool = True, output_limit: int = _DEFAULT_OUTPUT_LIMIT, + separate_stderr: bool = False, ) -> Tuple[VerificationResult, ...]: """Run only pre-registered argv commands; never evaluate a model command string.""" if not isinstance(registry, VerificationRegistry): raise TypeError("registry must be VerificationRegistry") if type(stop_on_failure) is not bool: raise TypeError("stop_on_failure must be boolean") + if type(separate_stderr) is not bool: + raise TypeError('separate_stderr must be boolean') if isinstance(output_limit, bool) or not isinstance(output_limit, int) or not 1 <= output_limit <= 16 * 1024 * 1024: raise ValueError("output_limit must be between 1 byte and 16 MiB") root = _workspace_root(workspace) @@ -109,30 +112,36 @@ def run_verifications( results: list[VerificationResult] = [] for spec in specs: - scope = VerificationProcess(spec.argv, root) + scope = VerificationProcess(spec.argv, root, separate_stderr=True) if separate_stderr else VerificationProcess(spec.argv, root) process = scope.process captured = bytearray() + diagnostics = bytearray() + output_lock = threading.Lock() overflow = threading.Event() reader_errors = [] - def read_output(): + def read_output(stream, destination): try: while True: - chunk = process.stdout.read1(min(65536, output_limit + 1)) + chunk = stream.read1(min(65536, output_limit + 1)) if not chunk: break - remaining = output_limit - len(captured) - captured.extend(chunk[:remaining]) - if len(chunk) > remaining: - overflow.set() - break + with output_lock: + remaining = output_limit - len(captured) - len(diagnostics) + destination.extend(chunk[:remaining]) + if len(chunk) > remaining: + overflow.set() + break except BaseException as error: reader_errors.append(error) - reader = threading.Thread(target=read_output, daemon=True) + readers = [threading.Thread(target=read_output, args=(process.stdout,captured), daemon=True)] + if separate_stderr: + readers.append(threading.Thread(target=read_output, args=(process.stderr,diagnostics), daemon=True)) timed_out = False import time deadline = time.monotonic() + spec.timeout_seconds try: - reader.start() + for reader in readers: + reader.start() while process.poll() is None and not overflow.is_set(): remaining = deadline - time.monotonic() if remaining <= 0: @@ -144,10 +153,13 @@ def read_output(): try: # Also quiesce descendants after an otherwise successful parent. scope.terminate() - reader.join(timeout=3) - if reader.is_alive(): + for reader in readers: + reader.join(timeout=3) + if any(reader.is_alive() for reader in readers): raise RuntimeError('verification output reader did not become quiescent') process.stdout.close() + if separate_stderr: + process.stderr.close() finally: scope.close() if reader_errors: