diff --git a/README.md b/README.md index 94db057..7a3b669 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,5 @@ +현재 구현: [중단 가능한 Responses 스트림](docs/changes/guarded-responses-stream.md). 신규119/새전체9,435개(Python8,615+JS820)·정적 검사/빌드/OpenAPI·독립 검토를 통과했습니다. 모의 HTTP/SSE에서 공백 정체 후 닫힘·관측/최종 사용량·외부 취소를 검증했습니다. 이번 제공자 호출0이며 실제 제공자 중단/과금·기존 생성 SDK/앱 연결은 후속입니다. + 현재 구현: [JSON 연속 공백 감지기](docs/changes/json-whitespace-guard.md). 신규155/새전체9,316개(Python8,496+JS820)·정적 검사/빌드/OpenAPI·독립 검토를 통과했습니다. 기록된 v2 인자를 같은 위치에서 감지했지만 실제 요청 취소·앱 연결·비용 절감은 후속입니다. [compact 지침 진단](docs/research/question-evidence-stall-investigation.md)은1회/60.106초 타임아웃·응답/사용량 미관측·재호출0으로 효과 미판정입니다. 최신 실제 관측: [질문별 근거 생성 v2](docs/research/question-evidence-generation-v2-acceptance.md). 출력4,000상한에서도 미완료·SDK 거부였고, 응답 끝 연속 공백6,619bytes를 확인했습니다. 재호출 없이 사용량/실패를 보존하며 출력 정체 제어를 먼저 검토합니다. 의미·앱 인수는 남습니다. diff --git a/docs/README.md b/docs/README.md index cc223f9..000f9a7 100644 --- a/docs/README.md +++ b/docs/README.md @@ -1,3 +1,5 @@ +현재 구현: [중단 가능한 Responses 스트림](changes/guarded-responses-stream.md). 신규119/새전체9,435개(Python8,615+JS820)·정적 검사/빌드/OpenAPI·독립 검토를 통과했습니다. 모의 HTTP/SSE에서 공백 정체 후 닫힘·관측/최종 사용량·외부 취소를 검증했습니다. 이번 제공자 호출0이며 실제 제공자 중단/과금·기존 생성 SDK/앱 연결은 후속입니다. + 현재 구현: [JSON 연속 공백 감지기](changes/json-whitespace-guard.md). 신규155/새전체9,316개(Python8,496+JS820)·정적 검사/빌드/OpenAPI·독립 검토를 통과했습니다. 기록된 v2 인자를 같은 위치에서 감지했지만 실제 요청 취소·앱 연결·비용 절감은 후속입니다. [compact 지침 진단](research/question-evidence-stall-investigation.md)은1회/60.106초 타임아웃·응답/사용량 미관측·재호출0으로 효과 미판정입니다. 최신 실제 관측: [질문별 근거 생성 v2](research/question-evidence-generation-v2-acceptance.md). 출력4,000상한에서도 미완료·SDK 거부였고, 응답 끝 연속 공백6,619bytes를 확인했습니다. 재호출 없이 사용량/실패를 보존하며 출력 정체 제어를 먼저 검토합니다. 의미·앱 인수는 남습니다. diff --git a/docs/changes/guarded-responses-stream.md b/docs/changes/guarded-responses-stream.md new file mode 100644 index 0000000..d93832d --- /dev/null +++ b/docs/changes/guarded-responses-stream.md @@ -0,0 +1,56 @@ +# 중단 가능한 Responses 도구 스트림 + +2026-10-25 한국 시간. PR105의 순수 공백 감지기를 실제 OpenAI Python SDK 스트림 소비 경로에 연결하는 별도 재사용 transport다. OpenSpec root:null이므로 init 없이 SDD/TDD를 적용한다. 기존 앱/worker/생성 SDK와 그 영수증은 변경하지 않는다. 합성 HTTP/SSE 검증과 실제 제공자의 생성·과금·취소 보장은 구분한다. + +## Global Constraints + +Implementer owns ONLY new packages/agents/src/issuedigest_agents/guarded_responses_stream.py, new packages/agents/tests/test_guarded_responses_stream.py and Task1 checkbox. All470 existing non-Markdown execution files and prior experiment namespaces are frozen. Do not change exports/dependencies/schema/API/UI/server/DB/legacy SDK/runtime/model defaults/main. No private source data, provider calls, network, resource allocation, wholeQA, old runners or subagents. Controller owns fullQA/recorded-argument synthetic HTTP replay/docs/privacy/PR/Sites/cleanup. Borrow required caller-owned AsyncOpenAI; never create/close its client or change its settings. No retry/repair/fallback/truncation/background/retrieve/server-cancel requests. Stream close is local HTTP response close, not proof of provider compute cancellation/billing savings. + +## Spec + +New module public constants: GUARDED_RESPONSES_PROFILE="guarded-responses-tool-v1", STREAM_TOTAL_TOKEN_LIMIT=20000, STREAM_OUTPUT_TOKEN_LIMIT=4000, STREAM_DEADLINE_SECONDS=60, STREAM_CLOSE_DEADLINE_SECONDS=1, STREAM_ARGUMENT_BYTE_LIMIT=80000, STREAM_EVENT_LIMIT=10000. Public async function run_guarded_responses_tool(*, client:AsyncOpenAI, model:str, prompt:str, instructions:str, tool:ToolOutput[BaseModel])->GuardedResponsesToolResult. + +Result is frozen dataclass profile exact constant, output actual requested BaseModel type, usage:KnownUsage, usage_complete:Literal[True]. This is a transport result, NOT a stored generation receipt, source/access validation or semantic approval. Caller must run original source/question/canonical validation after using this transport; existing v2 receipts are not minted. + +Error GuardedResponsesError(ValueError) has safe fixed message "Guarded Responses request failed", known_usage:KnownUsage|None, may_have_started:bool, usage_complete:bool, stream_closed:bool, reason:Literal["input","whitespace","protocol","transport","timeout","usage","output","limit","cancelled"]. GuardedResponsesCancelled(asyncio.CancelledError) carries the same observations and fixed message so external cancellation remains cancellation. Raise from None, no raw event, input, partial output, original cause, raw provider IDs or error messages in exception text/extra fields. Properties only; don't claim adversarial deep immutability or traceback-frame scrubbing. Reason records primary failure even if closing fails; stream_closed only true when response is actually closed, never assumed. Unknown final usage must not be represented as zero/free. + +Preflight before dispatch: reject invalid/open-client state, client.max_retries other than exact integer0 (including bool), non-exact-str or blank model/prompt/instructions, invalid tool name (ASCII [A-Za-z0-9_-]{1,64}), nonexact ToolOutput instance, strict not True, output not a BaseModel class or config extra not forbid. Build schema with GenerateToolJsonSchema and OpenAIJsonSchemaTransformer(strict=True) matching pinned PydanticAI; reject errors. Use existing _request_envelope(prompt,[tool],instructions,4000)<=20000; preserve full strings and reserve actual normalized schema metadata. Nonblank means strip only for checking, never rewrite inputs. No arbitrary extra request kwargs. No client is created or closed. Invalid preflight is reason input, known_usage None/may_have_started False/usage_complete False/stream_closed False. + +One await client.responses.create with stream=True, store=False, background=False, parallel_tool_calls=False, model exact, instructions exact, input=prompt exact, reasoning={"effort":"low"}, service_tier="default", max_output_tokens=4000, timeout=60, tools=[normalized strict function with exact name/description/schema], tool_choice={"type":"function","name":tool.name}. Set may_have_started=True immediately before this await (uncertain dispatch is conservative). Wrap creation+consumption in asyncio.timeout(60), a new total operation deadline distinct from historical transport timeouts. Exactly one native HTTP request; SDK retries preflight-disabled. No SDK run loops or second calls. The caller continues to own its AsyncOpenAI/httpx lifetime. + +Use raw typed SSE events from the native SDK, before any cumulative argument assembly. Track strictly increasing exact nonnegative int sequence_number, one response.created (matching nonempty response ID) before other events, response.in_progress consistent ID/status if present. Only reasoning and exactly one expected function_call output item permitted. Track unique nonnegative output indexes/item IDs (bool invalid). Function identity includes name/id/call_id/output_index, cannot change; arguments initial string at output_item.added may be nonempty and must be guarded. Reasoning IDs/indexes cannot collide. Reasoning delta/summary events are ignored only for reasoning-related event type prefixes; they still count toward sequence/event limits and cannot introduce function args. Unknown/text/refusal/error/builtin tool events fail closed. Do not keep reasoning text. + +For response.function_call_arguments.delta: existing live function item/index must match; exact-str delta feeds JsonWhitespaceGuard before appending/using any of that chunk; a rejected chunk appends no prefix. Maintain one guard for that argument stream, cumulative accepted UTF-8 bytes<=80000. Empty deltas preserve lexical state. At threshold failure stop iterating immediately, close the HTTP response, do not drain future usage/events and return no partial result. Never strip/repair args. Guard any full argument snapshots before inspecting/using them (fresh guard for snapshot), then require exact equality to assembled delta text so done/terminal-only replacement cannot bypass guard. A function arguments.done and output_item.done must each match name/identity/arguments where supplied, occur once, and precede successful completion. A final function item has status completed; reasoning items can be final completed or status omitted as allowed by SDK. Reject extra/duplicate/out-of-order calls/items/deltas, contradictory snapshots or function completion before args.done. + +Terminal response.completed/incomplete/failed: response ID must match created ID. First validate/capture valid native usage even if later output/status checks fail. Native usage exact nonnegative integer input_tokens/output_tokens/total_tokens with total=input+output, requests=1, monotonically nondecreasing against any valid earlier observations. Preserve last valid observed counters on invalid/decreasing/missing usage. Observed counters above output4000/total20000 must be retained on reason usage failure, never clipped. usage_complete becomes True ONLY when a matching terminal event has valid usage. It means final usage observed, NOT generation complete. Earlier response.created/in_progress valid usage can be retained but usage_complete False. Missing/invalid terminal usage is failure; no successful free/default usage. Terminal incomplete/failed (and contradictory event/status) never succeeds even with parseable JSON. Completed success requires status completed/no error/incomplete_details, exactly one expected completed function plus permitted reasoning only, full identity/arguments agreement with accumulated stream/done snapshots, and all item completions. EOF without terminal is failure/unknown final usage. A terminal response defines the end: close without draining further events. + +On completed success only: parse exact assembled JSON without duplicate keys/nonfinite constants or numbers that overflow to nonfinite Python floats (recursively across lists/objects), require root object, validate requested output strictly, then existing _canonical_models/_actual_models. Invalid complete output retains terminal usage and fails reason output. No partial JSON parse or result before terminal. On protocol/whitespace/usage/limits/transport/deadline/external cancellation close this response in finally, preserving primary error and known usage. Bound cooperative close of the ACQUIRED HTTP response body with asyncio.timeout(STREAM_CLOSE_DEADLINE_SECONDS=1), including native SDK generator-finally close as well as our final cleanup; a narrowly owned AsyncByteStream proxy may wrap that response body before iteration without mutating borrowed client/settings. Use shared confirmed-close state/idempotence, never report closure merely from native is_closed. Pre-acquisition borrowed SDK internals are opaque/cooperative; this is not a hard end-to-end SLA. Bound cooperative finally-close separately with the same close policy; expiry must retain any earlier primary failure, stream_closed False unless closure confirmed, and no false cancellation classification from the cleanup timer. If there was no earlier error use reason timeout. Never leave an unowned background cleanup task; arbitrary cancellation-resistant custom closers cannot be forcibly stopped by Python, and this is not a hard wall-clock SLA. Creation/consumption may take60s plus up to1s cooperative cleanup and scheduling. External CancelledError propagates as GuardedResponsesCancelled, never ordinary success/ValueError. Closing failures must not leak provider messages; prevent a false stream_closed claim. Max10000 decoded events, max80000 accepted argument UTF-8 bytes and60-second operation deadline are explicit safety bounds; they do not cap decoder memory for a single native event or prove server-side stop time/cost. No overall app budget/reservation mutation. + +## Task 1: Guarded native stream transport + +- [x] Implement only new module/tests+checkbox with RED→GREEN TDD. + +Read test-driven-development. Use real AsyncOpenAI with max_retries=0 and httpx.MockTransport + close-observing AsyncByteStream supplying synthetic SSE. No provider/network/private fixtures. Spy wire body to assert all exact options/full prompt/instructions/strict schema/tool choice/one HTTP call; preflight rejects make0HTTPcalls. Successful complete typed output and usage; whole/1char/mixed/empty argument chunks, quoted long whitespace/Unicode unchanged. 256 vs257 outside whitespace; reject added/done/terminal snapshots with excess whitespace/mismatches; no suffix events consumed after guard failure; HTTP stream closed and borrowed client stays open. Valid observed nonterminal usage survives failure but is incomplete; no usage None neverzero; final usage preserved on invalidJSON/incomplete/overbudget; invalid/negative/bool/missing/inconsistent/decreasing usage cases. EOF, unknown event, reordered/wrong IDs/indexes/sequence/call name, second tools/text/refusal/errors, duplicates, missing done markers; schema output strictness/extra/wrongtype/duplicateJSONkeys/nonfinite; event/UTF8 bounds. Timeout/transport close errors and external cancellation tests use deterministic events or patched deadline (no60s waits); cancelled remains BaseException cancellation and carries usage. Sanitized fixed errors/cause, primary error/closed observations; no raw source values in repr/str. Strict focused pytest/scoped Ruff+format/mypy/diff only, no fullQA. Commit only owned3paths, write exact RED/GREEN/static/selfreview evidence in task report. No copied scanner oracle or tests that only mirror implementation. + +## Controller gates + +- [x] Independent task/wholebranch review C0I0M0·새전체9,435개(Python8,615+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0. +- [x] Old recorded v1/v2 arguments through new mocked native HTTP/SSE path, no provider; preserve470/priorproof/unknown paid usage. +- [x] Docs/privacy/links·Draft PR106 basePR105·Sitesv128/previousv127·7exact changed documents·ownQA3/volume3/SDD cleanup; no app/worker/main activation. + +## 판단과 비용 + +1. 별도 native streaming transport를 추가하고 기존 SDK/앱은 유지한다. 기존 영수증을 위조하지 않는 대신 앱의 실제 응답은 아직 바뀌지 않으며 후속 생성 SDK·worker 연결 인수가 필요하다. +2. 공백 감지 시 drain 없이 HTTP 응답을 닫고 관측 사용량/최종 여부를 나눈다. 알려진 비용을 보존하지만 제공자의 실제 중단·최종 과금은 미확인이며 비용 절감을 보장하지 않는다. +3. 새 경로에60초 전체 deadline·8만 argument UTF-8bytes·1만 event 제한을 명시한다. 무한 소비를 제한하지만 정상적인 긴 실행을 차단할 수 있고 단일 native event의 decoder 메모리는 별도다. + +4. 정리 중 외부 취소도 CancelledError 하위 클래스로 전파하되 먼저 발생한 실패 reason을 보존한다. reason만 보는 호출자는 취소를 놓칠 수 있으므로 예외 종류를 먼저 처리해야 한다. + +5. 정리에 별도1초 cooperative deadline을 둔다. 정상적인 느린 close도 미확인으로 실패할 수 있으며 Python 취소를 무시하는 임의 closer까지 강제 종료하지는 못한다. +6. JSON 숫자의 Python float overflow도 거부한다. inf 성공을 막는 대신 큰 수를 정확히 표현할 수 있는 다른 숫자 모델에도 문자열 등 별도 표현이 필요할 수 있다. + +공식 API 이벤트 기준: [Responses streaming consumers](https://developers.openai.com/api/docs/guides/migrate-to-responses#7-update-streaming-consumers). 실제 배포된 SDK는 로컬 잠금 버전 코드로 확인하며 현재 제공자 동작·요금은 별도 인수한다. + +## 검증 결과 + +실행1d1d0c7·합성119/새전체9,435개(Python8,615+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0, 태스크/전체 검토 C0I0M0다. [검증 기록](../research/guarded-responses-stream-validation.md)과 [Draft PR106](https://github.com/DigestWorks/IssueDigest/pull/106)을 따른다. 실제 제공자 호출0·기존470실행/과거 증거 보존이며 기존 SDK/worker 연결은 후속이다. diff --git a/docs/planning/implementation-plan.md b/docs/planning/implementation-plan.md index 1fdf5e2..4e58c83 100644 --- a/docs/planning/implementation-plan.md +++ b/docs/planning/implementation-plan.md @@ -2,6 +2,10 @@ 2026-10-20 KST 기준. IssueDigest는 질문에서 시작해 이슈의 맥락·쟁점·변화를 이해하는 B2C 브리핑 서비스다. [D016·D017·D019·D020·D021·D022](decisions.md)과 [제품 경험 인터뷰](../research/product-experience-interview.md)가 제품 방향의 원본이다. 이 문서는 현재 구현과 다음 구현을 구분한다. 두 사람의 공동 자료 검증을 최종 제품 구조로 확대하지 않는다. +## M3 현재 구현 · 스트림 transport와 생성 연결 + +[새 native 스트림](../changes/guarded-responses-stream.md)에 공백 감지기를 연결해 신규119/새전체9,435개(Python8,615+JS820)·독립 검토 C0I0M0를 통과했다. 모의 HTTP8회로 기존 미완료 인자의 닫힘/사용량을 재생했고 제공자 호출0이다. 관측 사용량과 최종 사용량 수신 여부, 불확실한 송신/응답 닫힘·외부 취소를 구분한다. 신규 경로만60초 operation deadline·acquired body 별도1초 cooperative 정리·8만 argument UTF-8bytes·1만 events로 제한하며 기존 SDK/영수증/앱은 유지한다. 합성 HTTP 검증과 실제 제공자의 중단·과금/생성 품질은 구분한다. + ## M3 현재 구현 · 순수 공백 감지와 다음 통합 [독립 JSON 공백 감지기](../changes/json-whitespace-guard.md)는 신규155/새전체9,316개(Python8,496+JS820)·정적 검사/빌드/OpenAPI·독립 검토 C0I0M0를 통과했다. 기록된 v2 인자는 네 분할에서 같은 위치에 감지했으며 v1 미완료는 그대로 통과했다. 문자열 밖 ASCII 연속 공백256까지만 허용하는 고정 휴리스틱이며 JSON 검증기·실제 스트림 취소가 아니다. 기존468실행 파일과 유료 경로는 유지한다. [compact 지침 1회 진단](../research/question-evidence-stall-investigation.md)은60.106초 타임아웃으로 HTTP 응답/사용량이 없어 효과 미판정·비용 미관측이다. 이후 통합에서는 취소 후 최종 사용량이 없는 상태를 별도로 처리해야 한다. diff --git a/docs/planning/progress.md b/docs/planning/progress.md index 2700924..4899a35 100644 --- a/docs/planning/progress.md +++ b/docs/planning/progress.md @@ -1,3 +1,21 @@ +## 최종 인계 · Responses 스트림·사용량 관측 · 2026-10-25 한국 시간 + +[Draft PR106](https://github.com/DigestWorks/IssueDigest/pull/106)은 PR105 기반·미병합이다. 런타임ab7649a/테스트 격리1d1d0c7·신규119/전체9,435QA(Python8,615+JS820), 정적 검사/빌드/OpenAPI·실패/skip/warning0이다. 초기 태스크의 close/숫자 overflow와 전체 검토의 최초 오류 보존을 수정·재검토했고, 전체 실행에서 드러난 새 테스트 모듈 격리는 순서300개·좁은 검토로 보완했다. 최종 검토 C0I0M0·미해결0, 중단/실패 전체 실행은 성공으로 세지 않고 별도 보존한다. + +Sitesv128 public/native succeeded(2026-10-25T10:44:16.362351+00:00), source `9fe955ef5a95531070769f88d5e94cf964fd5a41`, saved `appgprj_6ac612dd42708191a7a0c8377be64d48~appgver_49e453ff6b7081918915c1f602258b15`, deployment `appgdep_6adddd73bf44819190c5b4f6f3a1da0c`다. 이전v127/변경7문서 정확 사본(be4c102)을 보존했고 HTML113/로컬대상3,143/공개텍스트339·비공개60+12항목 제외를 확인했다. 새브라우저QA/사용자탭 이동0이며 게시 이후 완료 기록은 저장소 원본을 따른다. + +기존470실행/과거 증거는 불변이고 단계 전체 제공자 호출0이다. 최종 코드 기준8회 모의 HTTP는 v2 조기닫힘/최종 usage 미관측과 v1 미완료/알려진 최종 usage를 보존했다. 실제 제공자의 계산 취소·과금 절감·의미·앱 인수로 확대하지 않는다. 자체QA3개/전용 익명볼륨3개·현재SDD만 정리하고25590닫힘·다른자원 변경0을 확인했다. 여섯 판단과 비용/검증을 명세·PR·비공개에 남겼으며 브랜치/격리폴더는 유지한다. 앱/worker/main 활성화0, 다음은 생성 SDK/worker 연결·실제 완료/의미·앱 인수다. 최종 문서 push의 원격CI는 로컬 검증과 구분한다. + +## 검증 완료 · 중단 가능한 Responses 스트림 · 2026-10-25 한국 시간 + +[Draft PR106](https://github.com/DigestWorks/IssueDigest/pull/106)은 PR105 기반·미병합이다. 실행1d1d0c7·신규119/새전체9,435개(Python8,615+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·태스크/전체 검토 C0I0M0·미해결0이다. 실제 native SDK+모의 HTTP에서 단일 요청/무재시도·전체 입력/strict schema·닫힘/외부 취소·관측/최종 사용량을 검증했다. 기존470실행/과거 증거는 유지한다. + +기록된 v1/v2 인자 각각 네 분할을 새 transport로 재생했다. v2는 공백 감지 후 terminal까지 읽지 않고 HTTP를 닫아 known_usage None/usage_complete False를 유지한다. v1 미완료는 관측된 최종 사용량을 보존하고 실패한다. 모의 HTTP8회/제공자 호출0이며 실제 과금 절감·계산 취소·의미 성공의 증거가 아니다. 생성 SDK/worker 연결·실제 완료/의미·앱 인수는 후속이고 앱/worker/main 활성화0이다. 자체QA3개/전용 익명볼륨3개를 정리하고25590포트 닫힘을 확인했다. Sites/현재SDD 정리를 마무리한다. + +## 진행 · 중단 가능한 Responses 스트림 · 2026-10-25 한국 시간 + +[고정 계약](../changes/guarded-responses-stream.md)으로 실제 native SDK의 스트리밍 소비·공백 감지 후 HTTP 닫힘·관측/최종 사용량·취소 전파를 별도 구현한다. 기존470실행/이전 증거를 고정하고 합성 TDD·독립 검토·새 전체QA·기록된 인자 재생을 진행한다. 이번 제공자 호출0이며 기존 사용량 미관측을0/무료로 바꾸지 않는다. 앱/worker/main 활성화와 실제 제공자 중단·과금/의미 인수는 별도다. + ## 최종 인계 · JSON 연속 공백 감지 · 2026-10-23 한국 시간 [Draft PR105](https://github.com/DigestWorks/IssueDigest/pull/105)은 PR104 기반·미병합이다. 실행63fc05b·전체9,316QA(Python8,496+JS820), 정적 검사/빌드/OpenAPI·실패/skip/warning0·태스크/전체 검토 C0I0M0·미해결0이다. compact 지침1회 진단은60.106초 타임아웃·응답/사용량 미관측·재호출0으로 유지한다. 순수 감지기의 기록 재생은 실제 취소·의미 성공·비용 절감이 아니다. diff --git a/docs/planning/tasks.md b/docs/planning/tasks.md index 39ea5c1..5fe16a8 100644 --- a/docs/planning/tasks.md +++ b/docs/planning/tasks.md @@ -1,3 +1,10 @@ +## 진행 · 중단 가능한 Responses 스트림 · 2026-10-25 한국 시간 + +- [x] [native transport 계약](../changes/guarded-responses-stream.md)·관측/최종 사용량·취소/응답 닫힘·명시적 한도 고정. +- [x] 새 모듈/합성 HTTP TDD119·정적 검사/빌드/OpenAPI·독립 검토 C0I0M0·새전체9,435개(Python8,615+JS820)·실패/skip/warning0. +- [x] [기존 인자 SSE 재생](../research/guarded-responses-stream-validation.md)·470실행/과거 증거 보존·문서/개인정보 검사·Draft PR106/Sitesv128·자체QA3/volume3/SDD 정리. +- [ ] 생성 SDK/worker 연결·실제 제공자 완료/의미/취소·과금·앱 인수. 기존 유료 경로는 유지. + ## 진행 · JSON 연속 공백 감지 · 2026-10-23 한국 시간 - [x] compact 지침 진단을 사전 고정하고1회 송신;60.106초 타임아웃·응답/사용량 미관측·재호출0 보존. diff --git a/docs/research/guarded-responses-stream-validation.md b/docs/research/guarded-responses-stream-validation.md new file mode 100644 index 0000000..36c961e --- /dev/null +++ b/docs/research/guarded-responses-stream-validation.md @@ -0,0 +1,35 @@ +# 스트림 중단·사용량 관측 검증 + +2026-10-25 한국 시간. [실행 계약](../changes/guarded-responses-stream.md)에 따른 별도 native Responses transport다. 이번 단계의 제공자 호출은0회이며 기존 실제 미완료·타임아웃/최종 사용량 미관측 기록과 소진 잠금은 유지한다. + +## 검증 구분 + +합성 HTTP/SSE는 설치된 OpenAI SDK를 실제 통과시키되 MockTransport에서 응답한다. 요청 body/strict schema/전체 입력·단일 호출·HTTP 응답 닫힘·borrowed client 유지와 상태 전이를 검증한다. 실제 제공자의 계산 취소·과금·생성 품질·네트워크 종료 시간을 측정한 것이 아니다. + +기존 v1/v2의 미완료 arguments는 읽기 전용으로 재생한다. 공백 감지 실패에서 terminal usage까지 drain하지 않으므로 관측 사용량이 없을 수 있으며 이를0/무료로 변환하지 않는다. v1의 공백 정체 없는 미완료는 terminal incomplete와 usage를 보존해야 한다. 합성 분할은 실제 SSE chunk나 타이밍을 복원하지 않는다. + +실행1d1d0c7·신규119/새전체9,435개(Python8,615+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·태스크/전체 검토 C0I0M0를 확인했다. 기존 SDK·영수증/원문/질문·앱/worker/main 활성화는 변경하지 않는다. 검증 결과가 나온 후 정확한 건수·한계를 추가한다. + +## 오프라인 재생 관측 + +새 transport의 native SDK+모의 SSE로 기존 v1/v2 인자를 각각 전체·1자·2자·혼합 분할했다.8요청 모두 HTTP 응답이 닫혔고 caller 소유 client는 열린 상태였다. v2는 네 경우 모두 whitespace 실패·terminal 미독해·known_usage None/usage_complete False이며 비용0으로 해석하지 않는다. v1은 네 경우 모두 미완료 실패·known_usage2,318/1,800·requests1/usage_complete True다. 후자는 최종 사용량 수신이지 생성 완료가 아니다. + +provider 요청0·원래 응답 SHA 불변이며, 실제 SSE 타이밍·제공자 계산 중단·과금 절감·모델 의미 성공은 검증하지 않았다. 재생 도구의 최초 입력 준비 오류는 tuple Source 구성으로 수정한 후 실행했으며 제공자 호출은 어느 경우에도 없었다. 새 실행 코드는 변경하지 않았다. + +## 독립 검토 보완 + +초기 검토에서 native/호출자 정리의 close 대기가 operation timer 밖에서 멈출 수 있는 문제와 JSON 지수 overflow가 inf 결과로 수용되는 경계를 재현했다. 기존 전체QA는 성공으로 세지 않고 중단/기록한 뒤 새 close 정책·finite 숫자 회귀를 보완한다. 취소에 협조하는 acquired HTTP body 정리는 별도1초 제한을 공유하며, native SDK 내부 정리도 같은 경계를 통과한다. 정리 취소를 무시하는 임의 closer·응답을 얻기 전의 borrowed SDK 내부까지 강제 종료하는 hard SLA는 아니다. + +수정 실행88c93ef·관련 합성115/정적 검사·양쪽 새 파일 strict mypy가 통과했다. 수정 후 동일한8회 모의 HTTP 재생에서도 v1 최종 usage/미완료와 v2 공백 감지/최종 usage 미관측/HTTP 닫힘을 보존했다. 실제 제공자 요청은 여전히0이며 독립 재검토·새 전체QA는 진행 중이다. + +## 기록된 인자의 native HTTP/SSE 재생 + +원래 응답 파일은 변경하지 않고 v1/v2 각각 전체·1자·2자·혼합 분할로8회 모의 HTTP 요청했다. v2 네 경우는 모두 whitespace 실패·HTTP 닫힘·terminal 미독해·known_usage None/usage_complete False다. 여기서 None은 무료가 아니라 최종 사용량 미관측이다. v1 네 경우는 incomplete 실패와 원래 입력2,318/출력1,800(추론 포함)·requests1을 보존하고 usage_complete True다. 이는 최종 사용량 수신이지 생성 완료가 아니다. borrowed client는 각 실행 후 열린 상태였다. + +제공자 요청·추가 retry·JSON 수리0이다. 합성 분할은 실제 SSE 타이밍이 아니며 HTTP 응답 닫힘으로 실제 제공자의 계산 중단/최종 과금/지연 절감을 입증하지 않는다. 과거 실패·사용량 미관측 기록을 성공이나 비용0으로 바꾸지 않는다. + +## 전체 실행의 테스트 격리 보완 + +완료된 초기 전체 실행에서1,492실패/6,918통과/205setup오류가 나타났다. 신규 테스트가 도입한 모듈이 frozen reload 이후 예전 Pydantic 클래스 참조를 유지한 테스트 격리 문제였다. 단독 기존 사례1개는 통과하고 순서 재현은1실패/299통과였다. 새 테스트 파일의 모듈/패키지 namespace 복원 fixture만 보완해 순서300개·focused119개와 좁은 독립 검토가 통과했다. 런타임ab7649a·기존470실행/강한 타입 검사는 바꾸지 않았다. + +최종 코드1d1d0c7의 fresh 전체는새전체9,435개(Python8,615+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0이다. 중단/실패 실행은 성공으로 세지 않고 별도 증거로 보존했다. 최종 코드 기준8회 모의 HTTP 재생과 이전 코드 재생 증거를 구분하며 제공자 요청은 단계 전체0회다. diff --git a/packages/agents/src/issuedigest_agents/guarded_responses_stream.py b/packages/agents/src/issuedigest_agents/guarded_responses_stream.py new file mode 100644 index 0000000..04af325 --- /dev/null +++ b/packages/agents/src/issuedigest_agents/guarded_responses_stream.py @@ -0,0 +1,539 @@ +"""Bounded native Responses tool transport; callers still validate source and meaning. + +Closing the local HTTP response does not establish provider cancellation or final cost. +This result is not a stored generation receipt and does not activate any app runtime. +""" + +import asyncio +import json +import math +import re +import sys +from collections.abc import AsyncIterator +from dataclasses import dataclass +from typing import Any, Literal, cast + +from httpx2 import AsyncByteStream +from openai import AsyncOpenAI, AsyncStream +from openai.types.responses import FunctionToolParam, ResponseStreamEvent +from pydantic import BaseModel +from pydantic_ai import ToolOutput +from pydantic_ai.profiles.openai import OpenAIJsonSchemaTransformer +from pydantic_ai.tools import GenerateToolJsonSchema + +from .json_whitespace_guard import JsonWhitespaceGuard, JsonWhitespaceGuardError +from .numeric_evidence import KnownUsage +from .numeric_selection_evaluation import _canonical_models +from .role_bounded_runtime import _request_envelope +from .shared_measurement_context import _actual_models + +GUARDED_RESPONSES_PROFILE: Literal["guarded-responses-tool-v1"] = "guarded-responses-tool-v1" +STREAM_TOTAL_TOKEN_LIMIT = 20000 +STREAM_OUTPUT_TOKEN_LIMIT = 4000 +STREAM_DEADLINE_SECONDS = 60 +STREAM_CLOSE_DEADLINE_SECONDS = 1 +STREAM_ARGUMENT_BYTE_LIMIT = 80000 +STREAM_EVENT_LIMIT = 10000 + +Reason = Literal[ + "input", + "whitespace", + "protocol", + "transport", + "timeout", + "usage", + "output", + "limit", + "cancelled", +] +_MESSAGE = "Guarded Responses request failed" + + +@dataclass(frozen=True) +class GuardedResponsesToolResult: + profile: Literal["guarded-responses-tool-v1"] + output: BaseModel + usage: KnownUsage + usage_complete: Literal[True] + + +class _Observations: + def __init__( + self, + *, + reason: Reason, + known_usage: KnownUsage | None = None, + may_have_started: bool = False, + usage_complete: bool = False, + stream_closed: bool = False, + ) -> None: + self._reason = reason + self._known_usage = known_usage + self._may_have_started = may_have_started + self._usage_complete = usage_complete + self._stream_closed = stream_closed + + @property + def reason(self) -> Reason: + return self._reason + + @property + def known_usage(self) -> KnownUsage | None: + return self._known_usage + + @property + def may_have_started(self) -> bool: + return self._may_have_started + + @property + def usage_complete(self) -> bool: + return self._usage_complete + + @property + def stream_closed(self) -> bool: + return self._stream_closed + + +class GuardedResponsesError(ValueError, _Observations): + def __init__(self, **observations: Any) -> None: + ValueError.__init__(self, _MESSAGE) + _Observations.__init__(self, **observations) + + +class GuardedResponsesCancelled(asyncio.CancelledError, _Observations): + def __init__(self, **observations: Any) -> None: + asyncio.CancelledError.__init__(self, _MESSAGE) + _Observations.__init__(self, **observations) + + +class _Failure(Exception): + def __init__(self, reason: Reason) -> None: + self.reason = reason + + +class _BoundedBody(AsyncByteStream): + """Own only an acquired response body; confirm one cooperative close attempt.""" + + def __init__(self, body: AsyncByteStream, operation_deadline: asyncio.Timeout) -> None: + self._body = body + self._operation_deadline = operation_deadline + self._attempted = False + self.confirmed_closed = False + self._failure: Reason | None = None + self.primary_reason: Reason | None = None + + async def __aiter__(self) -> AsyncIterator[bytes]: + async for chunk in self._body: + yield chunk + + async def aclose(self) -> None: + if self.confirmed_closed: + return + if self._attempted: + raise _Failure(self._failure or "transport") from None + self._attempted = True + # Native AsyncStream invokes close from its generator's finally. Preserve + # that already-active failure if body cleanup raises a different exception. + pending = sys.exception() + if isinstance(pending, asyncio.CancelledError): + self.primary_reason = "timeout" if self._operation_deadline.expired() else "cancelled" + elif isinstance(pending, TimeoutError): + self.primary_reason = "timeout" + elif isinstance(pending, Exception): + self.primary_reason = "transport" + close_deadline = asyncio.timeout(STREAM_CLOSE_DEADLINE_SECONDS) + try: + async with close_deadline: + await self._body.aclose() + self.confirmed_closed = True + if close_deadline.expired(): + raise _Failure("timeout") + except asyncio.CancelledError: + self._failure = self.primary_reason or "cancelled" + raise + except Exception: + close_reason: Reason = "timeout" if close_deadline.expired() else "transport" + self._failure = self.primary_reason or close_reason + raise _Failure(self._failure) from None + + +def _require(condition: bool, reason: Reason = "protocol") -> None: + if not condition: + raise _Failure(reason) + + +def _string(value: Any) -> bool: + return type(value) is str and bool(value) + + +def _index(value: Any) -> bool: + return type(value) is int and value >= 0 + + +def _snapshot(value: Any) -> str: + _require(type(value) is str) + JsonWhitespaceGuard().feed(value) + return cast(str, value) + + +class _State: + def __init__(self, name: str) -> None: + self.name = name + self.response_id: str | None = None + self.sequence = -1 + self.events = 0 + self.items: dict[int, tuple[str, str]] = {} + self.finished: set[int] = set() + self.function_index: int | None = None + self.call_id: str | None = None + self.arguments = "" + self.argument_bytes = 0 + self.guard = JsonWhitespaceGuard() + self.arguments_done = False + self.known_usage: KnownUsage | None = None + self.usage_complete = False + + def usage(self, usage: Any, *, terminal: bool) -> None: + if usage is None and not terminal: + return + _require(isinstance(usage, dict), "usage") + values = [usage.get(key) for key in ("input_tokens", "output_tokens", "total_tokens")] + _require(all(_index(value) for value in values), "usage") + inputs, outputs, total = values + _require(total == inputs + outputs, "usage") + prior = self.known_usage + _require( + prior is None or (inputs >= prior.input_tokens and outputs >= prior.output_tokens), + "usage", + ) + self.known_usage = KnownUsage(input_tokens=inputs, output_tokens=outputs, requests=1) + self.usage_complete = terminal + _require( + outputs <= STREAM_OUTPUT_TOKEN_LIMIT and total <= STREAM_TOTAL_TOKEN_LIMIT, "usage" + ) + + def append(self, text: Any) -> None: + _require(type(text) is str) + self.guard.feed(text) # A rejected chunk must never contribute even a prefix. + size = len(text.encode("utf-8")) + _require(self.argument_bytes + size <= STREAM_ARGUMENT_BYTE_LIMIT, "limit") + self.arguments += text + self.argument_bytes += size + + def identity(self, item: dict[str, Any], index: Any, *, final: bool) -> None: + _require(_index(index) and index == self.function_index) + _require(self.items.get(index) == (item.get("id"), "function_call")) + _require( + item.get("type") == "function_call" + and item.get("name") == self.name + and item.get("call_id") == self.call_id + ) + _require(_snapshot(item.get("arguments")) == self.arguments) + if final: + _require(item.get("status") == "completed") + + def consume(self, event: dict[str, Any]) -> bool: + response: Any + item: Any + index: Any + self.events += 1 + _require(self.events <= STREAM_EVENT_LIMIT, "limit") + sequence: Any = event.get("sequence_number") + _require(_index(sequence) and sequence > self.sequence) + self.sequence = sequence + kind = event.get("type") + if kind == "response.created": + _require(self.response_id is None) + response = event.get("response") + _require(isinstance(response, dict) and _string(response.get("id"))) + self.response_id = response["id"] + self.usage(response.get("usage"), terminal=False) + _require(response.get("status") == "in_progress" and not response.get("output")) + _require(response.get("error") is None and response.get("incomplete_details") is None) + return False + _require(self.response_id is not None) + if kind in ( + "response.in_progress", + "response.completed", + "response.incomplete", + "response.failed", + ): + response = event.get("response") + _require(isinstance(response, dict) and response.get("id") == self.response_id) + terminal = kind != "response.in_progress" + self.usage(response.get("usage"), terminal=terminal) + if not terminal: + _require(response.get("status") == "in_progress" and not response.get("output")) + _require( + response.get("error") is None and response.get("incomplete_details") is None + ) + return False + _require(kind == "response.completed" and response.get("status") == "completed") + _require(response.get("error") is None and response.get("incomplete_details") is None) + _require( + self.arguments_done + and self.function_index is not None + and self.finished == self.items.keys() + ) + output = response.get("output") + _require(type(output) is list and len(output) == len(self.items)) + for index, item in enumerate(output): + _require(isinstance(item, dict)) + _require(self.items.get(index) == (item.get("id"), item.get("type"))) + if item.get("type") == "function_call": + self.identity(item, index, final=True) + else: + _require( + item.get("type") == "reasoning" + and item.get("status") in (None, "completed") + ) + _require("arguments" not in item) + return True + if kind in ("response.output_item.added", "response.output_item.done"): + item = event.get("item") + index = event.get("output_index") + _require(isinstance(item, dict) and _index(index) and _string(item.get("id"))) + item_kind = item.get("type") + _require(item_kind in ("reasoning", "function_call")) + if kind.endswith(".added"): + _require( + index not in self.items + and item["id"] not in {v[0] for v in self.items.values()} + ) + if item_kind == "function_call": + _require( + self.function_index is None + and item.get("name") == self.name + and _string(item.get("call_id")) + and item.get("status") == "in_progress" + ) + self.function_index, self.call_id = index, item["call_id"] + self.append(item.get("arguments")) + else: + _require( + "arguments" not in item and item.get("status") in (None, "in_progress") + ) + self.items[index] = (item["id"], item_kind) + else: + _require( + index not in self.finished and self.items.get(index) == (item["id"], item_kind) + ) + if item_kind == "function_call": + _require(self.arguments_done) + self.identity(item, index, final=True) + else: + _require("arguments" not in item and item.get("status") in (None, "completed")) + self.finished.add(index) + return False + if kind in ( + "response.function_call_arguments.delta", + "response.function_call_arguments.done", + ): + index = event.get("output_index") + _require( + _index(index) + and index == self.function_index + and not self.arguments_done + and self.items.get(index) == (event.get("item_id"), "function_call") + ) + for key, expected in (("name", self.name), ("call_id", self.call_id)): + _require(key not in event or event[key] == expected) + if kind.endswith(".delta"): + self.append(event.get("delta")) + else: + _require(_snapshot(event.get("arguments")) == self.arguments) + self.arguments_done = True + return False + if type(kind) is str and kind.startswith( + ( + "response.reasoning_text.", + "response.reasoning_summary_text.", + "response.reasoning_summary_part.", + ) + ): + index = event.get("output_index") + _require( + _index(index) + and self.items.get(index) == (event.get("item_id"), "reasoning") + and index not in self.finished + ) + _require(not any(key in event for key in ("arguments", "delta_arguments", "call_id"))) + return False + raise _Failure("protocol") + + +def _pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + _require(key not in result, "output") + result[key] = value + return result + + +def _constant(value: str) -> Any: + raise _Failure("output") + + +def _finite_numbers(value: Any) -> None: + if type(value) is float: + _require(math.isfinite(value), "output") + elif isinstance(value, dict): + for child in value.values(): + _finite_numbers(child) + elif isinstance(value, list): + for child in value: + _finite_numbers(child) + + +def _prepare( + client: AsyncOpenAI, model: str, prompt: str, instructions: str, tool: ToolOutput[BaseModel] +) -> FunctionToolParam: + _require(isinstance(client, AsyncOpenAI) and not client.is_closed(), "input") + _require(type(client.max_retries) is int and client.max_retries == 0, "input") + _require( + all(type(value) is str and bool(value.strip()) for value in (model, prompt, instructions)), + "input", + ) + _require(type(tool) is ToolOutput and tool.strict is True, "input") + _require( + type(tool.name) is str and re.fullmatch(r"[A-Za-z0-9_-]{1,64}", tool.name) is not None, + "input", + ) + _require(tool.description is None or type(tool.description) is str, "input") + output = tool.output + _require(isinstance(output, type) and issubclass(output, BaseModel), "input") + output = cast(type[BaseModel], output) + _require(output.model_config.get("extra") == "forbid", "input") + raw = output.model_json_schema(schema_generator=GenerateToolJsonSchema) + parameters = OpenAIJsonSchemaTransformer(raw, strict=True).walk() + _require( + _request_envelope(prompt, [tool], instructions, STREAM_OUTPUT_TOKEN_LIMIT) + <= STREAM_TOTAL_TOKEN_LIMIT, + "input", + ) + native: FunctionToolParam = { + "type": "function", + "name": cast(str, tool.name), + "strict": True, + "parameters": parameters, + } + if tool.description is not None: + native["description"] = tool.description + return native + + +async def run_guarded_responses_tool( + *, client: AsyncOpenAI, model: str, prompt: str, instructions: str, tool: ToolOutput[BaseModel] +) -> GuardedResponsesToolResult: + """Consume one borrowed native request, rejecting unsafe arguments before assembly.""" + try: + native = _prepare(client, model, prompt, instructions, tool) + except Exception: + raise GuardedResponsesError(reason="input") from None + state = _State(native["name"]) + stream: AsyncStream[ResponseStreamEvent] | None = None + body: _BoundedBody | None = None + reason: Reason | None = None + output: BaseModel | None = None + stream_closed = False + cancelled = False + may_have_started = False + task = asyncio.current_task() + cancellations = task.cancelling() if task is not None else 0 + deadline = asyncio.timeout(STREAM_DEADLINE_SECONDS) + try: + async with deadline: + may_have_started = True + stream = await client.responses.create( + model=model, + input=prompt, + instructions=instructions, + tools=[native], + tool_choice={"type": "function", "name": native["name"]}, + stream=True, + store=False, + background=False, + parallel_tool_calls=False, + reasoning={"effort": "low"}, + service_tier="default", + max_output_tokens=STREAM_OUTPUT_TOKEN_LIMIT, + timeout=STREAM_DEADLINE_SECONDS, + ) + body = _BoundedBody(cast(AsyncByteStream, stream.response.stream), deadline) + stream.response.stream = body + terminal = False + async for event in stream: + if state.consume(event.model_dump(exclude_unset=True, warnings=False)): + terminal = True + break + _require(terminal) + try: + parsed = json.loads( + state.arguments, object_pairs_hook=_pairs, parse_constant=_constant + ) + _require(type(parsed) is dict, "output") + _finite_numbers(parsed) + output_type = cast(type[BaseModel], tool.output) + output = output_type.model_validate_json(state.arguments, strict=True) + _require(type(output) is output_type, "output") + _canonical_models(output, output_type) + _actual_models(output, output_type) + except Exception: + raise _Failure("output") from None + except asyncio.CancelledError: + cancelled = True + reason = body.primary_reason if body is not None and body.primary_reason else "cancelled" + except TimeoutError: + reason = "timeout" + except JsonWhitespaceGuardError: + reason = "whitespace" + except _Failure as exc: + reason = exc.reason + cancelled = reason == "cancelled" + if task is not None and task.cancelling() > cancellations: + cancelled = True + if reason == "timeout" and (body is None or body.primary_reason is None): + reason = "cancelled" + except Exception: + # The SDK closes in its generator's finally. A failed close may replace + # CancelledError; preserve the task's outstanding cancellation intent. + if task is not None and task.cancelling() > cancellations: + cancelled = True + reason = ( + body.primary_reason if body is not None and body.primary_reason else "cancelled" + ) + else: + reason = "timeout" if deadline.expired() else "transport" + finally: + if stream is not None: + close_deadline = asyncio.timeout(STREAM_CLOSE_DEADLINE_SECONDS) + try: + async with close_deadline: + # httpx marks is_closed before awaiting its body close. Confirm closure + # explicitly even if native EOF/error cleanup already set that flag. + if stream.response.is_closed: + await cast(AsyncByteStream, stream.response.stream).aclose() + else: + await stream.close() + stream_closed = ( + body is not None and body.confirmed_closed and stream.response.is_closed + ) + except asyncio.CancelledError: + cancelled = True + reason = reason or "cancelled" + except Exception: + if task is not None and task.cancelling() > cancellations: + cancelled = True + close_reason: Reason = "timeout" if close_deadline.expired() else "transport" + reason = reason or ("cancelled" if cancelled else close_reason) + if reason is not None: + error_type = GuardedResponsesCancelled if cancelled else GuardedResponsesError + raise error_type( + reason=reason, + known_usage=state.known_usage, + may_have_started=may_have_started, + usage_complete=state.usage_complete, + stream_closed=stream_closed, + ) from None + assert output is not None and state.known_usage is not None + return GuardedResponsesToolResult(GUARDED_RESPONSES_PROFILE, output, state.known_usage, True) diff --git a/packages/agents/tests/test_guarded_responses_stream.py b/packages/agents/tests/test_guarded_responses_stream.py new file mode 100644 index 0000000..1d47c69 --- /dev/null +++ b/packages/agents/tests/test_guarded_responses_stream.py @@ -0,0 +1,1019 @@ +"""Real native SDK over synthetic SSE; no provider or source inputs.""" + +import asyncio +import importlib +import importlib.util +import json +import sys +from collections.abc import AsyncIterator, Iterator +from copy import deepcopy +from dataclasses import FrozenInstanceError +from typing import Any, cast + +import httpx2 as httpx +import pytest +from openai import AsyncOpenAI +from pydantic import BaseModel, ConfigDict +from pydantic_ai import ToolOutput + + +@pytest.fixture(scope="module", autouse=True) +def restore_introduced_agent_modules() -> Iterator[None]: + package = importlib.import_module("issuedigest_agents") + existing = set(sys.modules) + attributes = vars(package).copy() + try: + yield + finally: + for name in set(sys.modules) - existing: + if name.startswith("issuedigest_agents."): + sys.modules.pop(name, None) + # Restore package attributes too: from-package imports can otherwise reuse + # unloaded helpers whose model classes precede later frozen reload tests. + for name in set(vars(package)) - attributes.keys(): + delattr(package, name) + for name, value in attributes.items(): + setattr(package, name, value) + + +class Output(BaseModel): + model_config = ConfigDict(extra="forbid") + text: str + + +def api() -> Any: + name = "issuedigest_agents.guarded_responses_stream" + assert importlib.util.find_spec(name) is not None, "Missing guarded native transport" + return importlib.import_module(name) + + +def item(arguments: str = '{"text":"ok"}', **changes: Any) -> dict[str, Any]: + return dict( + type="function_call", + id="fc_1", + call_id="call_1", + name="answer", + arguments=arguments, + status="completed", + **changes, + ) + + +def response(status: Any = "completed", arguments: Any = '{"text":"ok"}', usage: Any = None) -> Any: + return dict( + id="resp_1", + object="response", + created_at=1, + status=status, + error=None, + incomplete_details=None, + output=[item(arguments)], + usage=usage or dict(input_tokens=10, output_tokens=5, total_tokens=15), + ) + + +def events( + arguments: str = '{"text":"ok"}', chunks: list[str] | None = None, initial: str = "" +) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [ + dict( + type="response.created", + response={**response("in_progress"), "output": [], "usage": None}, + ), + dict( + type="response.output_item.added", + output_index=0, + item={**item(initial), "status": "in_progress"}, + ), + ] + rows += [ + dict( + type="response.function_call_arguments.delta", + output_index=0, + item_id="fc_1", + delta=chunk, + ) + for chunk in ([arguments] if chunks is None else chunks) + ] + rows += [ + dict( + type="response.function_call_arguments.done", + output_index=0, + item_id="fc_1", + arguments=arguments, + ), + dict(type="response.output_item.done", output_index=0, item=item(arguments)), + dict(type="response.completed", response=response(arguments=arguments)), + ] + return numbered(rows) + + +def numbered(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + return [{**row, "sequence_number": i} for i, row in enumerate(rows)] + + +class Body(httpx.AsyncByteStream): + def __init__( + self, + rows: list[dict[str, Any]], + *, + wait: bool = False, + close_error: bool = False, + read_error: bool = False, + ) -> None: + self.rows = rows + self.consumed = 0 + self.closed = False + self.wait = wait + self.close_error = close_error + self.read_error = read_error + self.blocked = asyncio.Event() + + async def __aiter__(self) -> AsyncIterator[bytes]: + for row in self.rows: + self.consumed += 1 + yield ("event: " + row["type"] + "\ndata: " + json.dumps(row) + "\n\n").encode() + if self.read_error: + raise httpx.ReadError("sensitive transport details") + if self.wait: + self.blocked.set() + await asyncio.Event().wait() + + async def aclose(self) -> None: + if self.close_error: + raise RuntimeError("sensitive close details") + self.closed = True + + +async def harness( + rows: list[dict[str, Any]] | None = None, + *, + body: Body | None = None, + kwargs: dict[str, Any] | None = None, + client_kwargs: dict[str, Any] | None = None, + cancel: bool = False, +) -> tuple[Any, Body, list[httpx.Request]]: + mod = api() + body = body or Body(events() if rows is None else rows) + requests: list[httpx.Request] = [] + + def serve(request: Any) -> Any: + requests.append(request) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, stream=body) + + http = httpx.AsyncClient(transport=httpx.MockTransport(serve)) + client_options: dict[str, Any] = dict( + api_key="synthetic", + base_url="https://synthetic.invalid/v1", + http_client=http, + max_retries=0, + ) + client_options.update(client_kwargs or {}) + client = AsyncOpenAI(**client_options) + options = dict( + client=client, + model=" model ", + prompt=" full prompt ", + instructions=" full instructions ", + tool=ToolOutput(Output, name="answer", description=" exact description ", strict=True), + ) + options.update(kwargs or {}) + try: + try: + if cancel: + task = asyncio.create_task(mod.run_guarded_responses_tool(**options)) + await body.blocked.wait() + task.cancel() + result = await task + else: + result = await mod.run_guarded_responses_tool(**options) + except BaseException as exc: + result = exc + assert not client.is_closed() + assert not http.is_closed + return (result, body, requests) + finally: + await client.close() + + +def run(*args: Any, **kwargs: Any) -> tuple[Any, Body, list[httpx.Request]]: + return asyncio.run(harness(*args, **kwargs)) + + +def failure( + result: Any, + reason: str, + *, + known: dict[str, int] | None = None, + complete: bool = False, + closed: bool = True, + started: bool = True, +) -> None: + mod = api() + assert isinstance(result, mod.GuardedResponsesError) + assert str(result) == "Guarded Responses request failed" + assert "sensitive" not in repr(result) + assert result.__cause__ is None + assert result.__suppress_context__ + assert result.reason == reason + assert result.usage_complete is complete + assert result.stream_closed is closed + assert result.may_have_started is started + if known is None: + assert result.known_usage is None + else: + assert result.known_usage.model_dump() == known + + +USAGE = dict(input_tokens=10, output_tokens=5, requests=1) + + +@pytest.mark.parametrize("chunks", [None, list('{"text":"ok"}'), ["", '{"text":', "", '"ok"}', ""]]) +def test_complete_native_request_preserves_wire_options_and_borrowed_client(chunks: Any) -> None: + result, body, requests = run(events(chunks=chunks)) + assert isinstance(result, api().GuardedResponsesToolResult) + assert type(result.output) is Output + assert result.output.text == "ok" + assert result.usage.model_dump() == USAGE + assert result.usage_complete is True + assert result.profile == "guarded-responses-tool-v1" + with pytest.raises(FrozenInstanceError): + result.profile = "changed" + assert body.closed + assert len(requests) == 1 + wire = json.loads(requests[0].content) + assert wire == dict( + model=" model ", + input=" full prompt ", + instructions=" full instructions ", + stream=True, + store=False, + background=False, + parallel_tool_calls=False, + reasoning={"effort": "low"}, + service_tier="default", + max_output_tokens=4000, + tool_choice={"type": "function", "name": "answer"}, + tools=[ + dict( + type="function", + name="answer", + description=" exact description ", + strict=True, + parameters={ + "additionalProperties": False, + "properties": {"text": {"type": "string"}}, + "required": ["text"], + "type": "object", + }, + ) + ], + ) + assert requests[0].extensions["timeout"]["read"] == 60 + + +@pytest.mark.parametrize( + "arguments", + [" " * 256 + '{"text":"ok"}', json.dumps({"text": " " * 300 + '한😀\\"'}, ensure_ascii=False)], +) +def test_lexical_guard_accepts_boundary_and_quoted_whitespace_unicode(arguments: Any) -> None: + result, _, _ = run(events(arguments, chunks=list(arguments))) + assert isinstance(result, api().GuardedResponsesToolResult) + assert result.output.text == json.loads(arguments)["text"] + + +@pytest.mark.parametrize("where", ["delta", "added", "arguments_done", "item_done", "terminal"]) +def test_whitespace_in_any_argument_surface_stops_without_draining(where: Any) -> None: + rows = events() + bad = " " * 257 + '{"text":"ok"}' + index = {"delta": 2, "added": 1, "arguments_done": 3, "item_done": 4, "terminal": 5}[where] + if where == "delta": + rows[index]["delta"] = bad + elif where == "arguments_done": + rows[index]["arguments"] = bad + elif where == "terminal": + rows[index]["response"]["output"][0]["arguments"] = bad + else: + rows[index]["item"]["arguments"] = bad + result, body, _ = run(rows) + failure( + result, + "whitespace", + known=USAGE if where == "terminal" else None, + complete=where == "terminal", + ) + assert body.consumed == index + 1 + + +@pytest.mark.parametrize( + "options", + [ + dict(model=" "), + dict(prompt=""), + dict(instructions=1), + dict(tool=ToolOutput(Output, name="bad.name", strict=True)), + dict(tool=ToolOutput(Output, name="answer", strict=False)), + dict(client=None), + ], +) +def test_invalid_preflight_never_dispatches(options: Any) -> None: + result, body, requests = run(kwargs=options) + failure(result, "input", closed=False, started=False) + assert not requests + assert not body.closed + + +@pytest.mark.parametrize( + "mutation", + [ + "sequence_bool", + "sequence_repeat", + "wrong_id", + "wrong_index", + "wrong_item", + "wrong_name", + "wrong_call", + "unknown", + "text", + "refusal", + "second_tool", + "duplicate_done", + "missing_args_done", + "missing_item_done", + "snapshot", + "early_delta", + "created_twice", + "wrong_terminal_id", + "terminal_status", + "terminal_error", + "reasoning_collision", + ], +) +def test_protocol_identity_order_and_completion_are_fail_closed(mutation: Any) -> None: + rows = events() + terminal_usage = False + if mutation == "sequence_bool": + rows[2]["sequence_number"] = True + elif mutation == "sequence_repeat": + rows[2]["sequence_number"] = 1 + elif mutation == "wrong_id": + rows.insert( + 1, + dict(type="response.in_progress", response={**response("in_progress"), "id": "wrong"}), + ) + elif mutation == "wrong_index": + rows[2]["output_index"] = True + elif mutation == "wrong_item": + rows[2]["item_id"] = "wrong" + elif mutation == "wrong_name": + rows[1]["item"]["name"] = "wrong" + elif mutation == "wrong_call": + rows[4]["item"]["call_id"] = "wrong" + elif mutation in ("unknown", "text", "refusal"): + rows[2]["type"] = { + "unknown": "response.future", + "text": "response.output_text.delta", + "refusal": "response.refusal.delta", + }[mutation] + elif mutation == "second_tool": + rows.insert( + 2, + dict(type="response.output_item.added", output_index=1, item={**item(), "id": "fc_2"}), + ) + elif mutation == "duplicate_done": + rows.insert(4, deepcopy(rows[3])) + elif mutation == "missing_args_done": + del rows[3] + terminal_usage = False + elif mutation == "missing_item_done": + del rows[4] + terminal_usage = True + elif mutation == "snapshot": + rows[3]["arguments"] = '{"text":"changed"}' + elif mutation == "early_delta": + rows[1], rows[2] = (rows[2], rows[1]) + elif mutation == "created_twice": + rows.insert(1, deepcopy(rows[0])) + elif mutation == "wrong_terminal_id": + rows[-1]["response"]["id"] = "wrong" + elif mutation == "terminal_status": + rows[-1]["response"]["status"] = "incomplete" + terminal_usage = True + elif mutation == "terminal_error": + rows[-1]["response"]["error"] = {"code": "x", "message": "sensitive"} + terminal_usage = True + elif mutation == "reasoning_collision": + rows.insert( + 2, + dict( + type="response.output_item.added", + output_index=0, + item=dict(type="reasoning", id="r1", summary=[]), + ), + ) + if mutation not in ("sequence_bool", "sequence_repeat"): + rows = numbered(rows) + result, _, _ = run(rows) + failure(result, "protocol", known=USAGE if terminal_usage else None, complete=terminal_usage) + + +@pytest.mark.parametrize( + "arguments", + [ + '{"text":3}', + '{"text":"ok","extra":1}', + '{"text":"ok","text":"duplicate"}', + '{"text":NaN}', + '{"text":Infinity}', + "[]", + "{broken", + ], +) +def test_invalid_complete_output_retains_final_usage(arguments: Any) -> None: + result, _, _ = run(events(arguments)) + failure(result, "output", known=USAGE, complete=True) + + +@pytest.mark.parametrize( + "usage", + [ + None, + {}, + dict(input_tokens=-1, output_tokens=5, total_tokens=4), + dict(input_tokens=True, output_tokens=5, total_tokens=6), + dict(input_tokens=10, output_tokens=5, total_tokens=16), + dict(input_tokens="10", output_tokens=5, total_tokens=15), + ], +) +def test_bad_terminal_usage_never_becomes_free(usage: Any) -> None: + rows = events() + rows[-1]["response"]["usage"] = usage + result, _, _ = run(rows) + failure(result, "usage") + + +@pytest.mark.parametrize("status", ["incomplete", "failed"]) +def test_terminal_failure_retains_final_usage(status: Any) -> None: + rows = events() + rows[-1]["type"] = "response." + status + rows[-1]["response"]["status"] = status + result, _, _ = run(rows) + failure(result, "protocol", known=USAGE, complete=True) + + +def test_observed_usage_survives_whitespace_but_is_not_final() -> None: + rows = events() + rows[0]["response"]["usage"] = dict(input_tokens=10, output_tokens=5, total_tokens=15) + rows[2]["delta"] = " " * 257 + result, _, _ = run(rows) + failure(result, "whitespace", known=USAGE) + + +def test_decreasing_usage_retains_last_valid_observation() -> None: + rows = events() + rows[0]["response"]["usage"] = dict(input_tokens=11, output_tokens=5, total_tokens=16) + result, _, _ = run(rows) + failure(result, "usage", known=dict(input_tokens=11, output_tokens=5, requests=1)) + + +@pytest.mark.parametrize("inputs,outputs", [(10, 4001), (20000, 1)]) +def test_overbudget_final_usage_is_not_clipped(inputs: Any, outputs: Any) -> None: + rows = events() + rows[-1]["response"]["usage"] = dict( + input_tokens=inputs, output_tokens=outputs, total_tokens=inputs + outputs + ) + result, _, _ = run(rows) + failure( + result, + "usage", + known=dict(input_tokens=inputs, output_tokens=outputs, requests=1), + complete=True, + ) + + +def test_eof_without_terminal_never_returns_partial_output() -> None: + result, _, _ = run(events()[:-1]) + failure(result, "protocol") + + +def test_transport_failure_is_sanitized_and_closed() -> None: + result, _, _ = run(body=Body(events()[:1], read_error=True)) + failure(result, "transport") + + +def test_close_failure_does_not_replace_primary_whitespace_failure() -> None: + rows = events() + rows[2]["delta"] = " " * 257 + result, body, _ = run(body=Body(rows, close_error=True)) + failure(result, "whitespace", closed=False) + assert not body.closed + + +def test_external_cancellation_remains_cancellation_and_retains_observations() -> None: + rows = events()[:1] + rows[0]["response"]["usage"] = dict(input_tokens=10, output_tokens=5, total_tokens=15) + result, body, _ = run(body=Body(rows, wait=True), cancel=True) + assert isinstance(result, api().GuardedResponsesCancelled) + assert issubclass(type(result), asyncio.CancelledError) + assert not isinstance(result, Exception) + assert result.reason == "cancelled" + assert result.known_usage.model_dump() == USAGE + assert result.usage_complete is False + assert result.stream_closed and body.closed + assert str(result) == "Guarded Responses request failed" + + +def test_deadline_closes_without_waiting_sixty_seconds(monkeypatch: Any) -> None: + monkeypatch.setattr(api(), "STREAM_DEADLINE_SECONDS", 0.01) + result, _, _ = run(body=Body(events()[:1], wait=True)) + failure(result, "timeout") + + +def test_utf8_bound_and_event_count_bound(monkeypatch: Any) -> None: + monkeypatch.setattr(api(), "STREAM_ARGUMENT_BYTE_LIMIT", 14) + result, _, _ = run(events('{"text":"😀"}')) + failure(result, "limit") + monkeypatch.setattr(api(), "STREAM_EVENT_LIMIT", 2) + result, body, _ = run() + failure(result, "limit") + assert body.consumed == 3 + + +def test_cancellation_stays_cancellation_when_native_finally_close_fails() -> None: + result, body, _ = run(body=Body(events()[:1], wait=True, close_error=True), cancel=True) + assert isinstance(result, api().GuardedResponsesCancelled) + assert result.reason == "cancelled" + assert result.stream_closed is False + assert not body.closed + + +def test_error_observations_are_readonly_properties() -> None: + result, _, _ = run(events()[:-1]) + for name in ("reason", "known_usage", "may_have_started", "usage_complete", "stream_closed"): + with pytest.raises(AttributeError): + setattr(result, name, None) + + +def test_prior_suppressed_cancellation_does_not_reclassify_transport_failure() -> None: + + async def scenario() -> Any: + task = asyncio.current_task() + assert task is not None + task.cancel() + try: + await asyncio.sleep(0) + except asyncio.CancelledError: + pass + return await harness(body=Body(events()[:1], read_error=True)) + + result, _, _ = asyncio.run(scenario()) + failure(result, "transport") + + +def test_deadline_preserved_when_native_finally_close_fails(monkeypatch: Any) -> None: + monkeypatch.setattr(api(), "STREAM_DEADLINE_SECONDS", 0.01) + result, _, _ = run(body=Body(events()[:1], wait=True, close_error=True)) + failure(result, "timeout", closed=False) + + +@pytest.mark.parametrize("retries", [True, False, 1, 2]) +def test_preflight_disables_sdk_retry_before_any_http(retries: Any) -> None: + result, _, requests = run(client_kwargs=dict(max_retries=retries)) + failure(result, "input", closed=False, started=False) + assert not requests + + +@pytest.mark.parametrize( + "tool", + [ + ToolOutput(Output, name="answer", strict=cast(Any, 1)), + ToolOutput(Output, name="", strict=True), + ToolOutput(Output, name="한글", strict=True), + ToolOutput(Output, name="a" * 65, strict=True), + ToolOutput(str, name="answer", strict=True), + ToolOutput(cast(Any, Output(text="example")), name="answer", strict=True), + ], +) +def test_preflight_tool_identity_and_output_class_are_required(tool: Any) -> None: + result, _, requests = run(kwargs=dict(tool=tool)) + failure(result, "input", closed=False, started=False) + assert not requests + + +def test_preflight_requires_extra_forbid_and_exact_tool_type() -> None: + + class Loose(BaseModel): + text: str + + class DerivedTool(ToolOutput[BaseModel]): + pass + + for tool in ( + ToolOutput(Loose, name="answer", strict=True), + DerivedTool(Output, name="answer", strict=True), + ): + result, _, requests = run(kwargs=dict(tool=tool)) + failure(result, "input", closed=False, started=False) + assert not requests + + +def test_preflight_bounds_full_prompt_without_truncation() -> None: + result, _, requests = run(kwargs=dict(prompt="x" * 20000)) + failure(result, "input", closed=False, started=False) + assert not requests + + +def test_none_description_is_omitted_from_native_tool() -> None: + result, _, requests = run(kwargs=dict(tool=ToolOutput(Output, name="answer", strict=True))) + assert isinstance(result, api().GuardedResponsesToolResult) + assert "description" not in json.loads(requests[0].content)["tools"][0] + + +def test_nonempty_added_arguments_join_deltas_without_rewriting() -> None: + arguments = '{"text":"한😀"}' + rows = events(arguments, initial=arguments[:9], chunks=["", arguments[9:], ""]) + result, _, _ = run(rows) + assert result.output.text == "한😀" + + +def test_whitespace_run_continues_from_added_into_first_delta() -> None: + rows = events(initial=" " * 256, chunks=[" ", '{"text":"ok"}']) + result, body, _ = run(rows) + failure(result, "whitespace") + assert body.consumed == 3 + + +@pytest.mark.parametrize("where", ["item_done", "terminal"]) +def test_complete_snapshots_cannot_replace_stream_arguments(where: Any) -> None: + rows = events() + if where == "terminal": + rows[-1]["response"]["output"][0]["arguments"] = '{"text":"replacement"}' + else: + rows[4]["item"]["arguments"] = '{"text":"replacement"}' + result, _, _ = run(rows) + failure( + result, + "protocol", + known=USAGE if where == "terminal" else None, + complete=where == "terminal", + ) + + +def test_reasoning_is_permitted_and_terminal_stops_without_suffix_drain() -> None: + rows = events() + for row in rows: + if "output_index" in row: + row["output_index"] = 1 + reasoning = dict(type="reasoning", id="r1", summary=[]) + rows[1:1] = [ + dict(type="response.output_item.added", output_index=0, item=reasoning), + dict( + type="response.reasoning_summary_text.delta", + output_index=0, + item_id="r1", + summary_index=0, + delta="discard this reasoning", + ), + dict(type="response.output_item.done", output_index=0, item=reasoning), + ] + rows[-1]["response"]["output"].insert(0, reasoning) + rows.append(dict(type="response.output_text.delta", delta="must not consume")) + result, body, _ = run(numbered(rows)) + assert isinstance(result, api().GuardedResponsesToolResult) + assert body.consumed == len(rows) - 1 + + +@pytest.mark.parametrize( + "usage", + [ + dict(input_tokens=-1, output_tokens=5, total_tokens=4), + dict(input_tokens=10, output_tokens=True, total_tokens=11), + dict(input_tokens=10, output_tokens=5, total_tokens=17), + ], +) +def test_malformed_nonterminal_observation_retains_prior_counters(usage: Any) -> None: + rows = events() + rows[0]["response"]["usage"] = dict(input_tokens=10, output_tokens=5, total_tokens=15) + rows.insert( + 1, dict(type="response.in_progress", response={**response("in_progress"), "usage": usage}) + ) + result, _, _ = run(numbered(rows)) + failure(result, "usage", known=USAGE) + + +def test_successful_terminal_with_close_failure_stays_failure_and_keeps_usage() -> None: + result, _, _ = run(body=Body(events(), close_error=True)) + failure(result, "transport", known=USAGE, complete=True, closed=False) + + +def test_http_failure_does_not_retry_or_leak_provider_details() -> None: + + async def scenario() -> Any: + calls: list[httpx.Request] = [] + + def serve(request: Any) -> Any: + calls.append(request) + return httpx.Response(500, json={"error": {"message": "sensitive provider payload"}}) + + async with httpx.AsyncClient(transport=httpx.MockTransport(serve)) as http: + client = AsyncOpenAI(api_key="synthetic", http_client=http, max_retries=0) + try: + await api().run_guarded_responses_tool( + client=client, + model="m", + prompt="p", + instructions="i", + tool=ToolOutput(Output, name="answer", strict=True), + ) + except api().GuardedResponsesError as exc: + failure(exc, "transport", closed=False) + else: + pytest.fail("HTTP error became a successful result") + assert len(calls) == 1 + assert not client.is_closed() + await client.close() + + asyncio.run(scenario()) + + +def test_in_progress_cannot_smuggle_untracked_output() -> None: + rows = events() + rows.insert( + 1, + dict( + type="response.in_progress", + response={ + **response("in_progress"), + "usage": None, + "output": [dict(type="message", id="msg_1", content=[])], + }, + ), + ) + result, _, _ = run(numbered(rows)) + failure(result, "protocol") + + +def test_closed_borrowed_client_is_rejected_without_dispatch() -> None: + + async def scenario() -> Any: + requests: list[httpx.Request] = [] + + def serve(request: Any) -> Any: + requests.append(request) + raise AssertionError("closed client dispatched") + + http = httpx.AsyncClient(transport=httpx.MockTransport(serve)) + client = AsyncOpenAI(api_key="synthetic", http_client=http, max_retries=0) + await client.close() + with pytest.raises(api().GuardedResponsesError) as caught: + await api().run_guarded_responses_tool( + client=client, + model="m", + prompt="p", + instructions="i", + tool=ToolOutput(Output, name="answer", strict=True), + ) + failure(caught.value, "input", closed=False, started=False) + assert not requests + + asyncio.run(scenario()) + + +def test_schema_failure_is_sanitized_before_dispatch() -> None: + + class BadSchema(Output): + @classmethod + def __get_pydantic_json_schema__(cls, *args: Any, **kwargs: Any) -> Any: + raise RuntimeError("sensitive schema metadata") + + result, _, requests = run(kwargs=dict(tool=ToolOutput(BadSchema, name="answer", strict=True))) + failure(result, "input", closed=False, started=False) + assert not requests + + +@pytest.mark.parametrize("size", [80000, 80001]) +def test_actual_utf8_limit_accepts_boundary_and_rejects_one_more_byte(size: Any) -> None: + arguments = '{"text":"' + "😀" * 19997 + "a" * (size - 80000 + 1) + '"}' + assert len(arguments.encode()) == size + result, _, _ = run(events(arguments)) + if size == 80000: + assert isinstance(result, api().GuardedResponsesToolResult) + else: + failure(result, "limit") + + +@pytest.mark.parametrize("mask_cancellation", [False, True]) +def test_external_cancellation_during_failure_cleanup_still_propagates_as_cancellation( + mask_cancellation: bool, +) -> None: + class ClosingBody(Body): + async def aclose(self) -> None: + self.blocked.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + if mask_cancellation: + raise RuntimeError("sensitive cleanup details") from None + raise + + rows = events() + rows[2]["delta"] = " " * 257 + result, _, _ = run(body=ClosingBody(rows), cancel=True) + assert isinstance(result, api().GuardedResponsesCancelled) + assert result.reason == "whitespace" + assert result.stream_closed is False + + +@pytest.mark.parametrize("number", ["1e999", "-1e999"]) +@pytest.mark.parametrize("location", ["scalar", "object", "array"]) +def test_exponent_overflow_is_rejected_before_strict_output(number: str, location: str) -> None: + class FloatOutput(BaseModel): + model_config = ConfigDict(extra="forbid") + text: float + + class NestedOutput(BaseModel): + model_config = ConfigDict(extra="forbid") + text: dict[str, float] + + class ArrayOutput(BaseModel): + model_config = ConfigDict(extra="forbid") + text: tuple[float, ...] + + value, output = { + "scalar": (number, FloatOutput), + "object": ('{"nested":' + number + "}", NestedOutput), + "array": ("[" + number + "]", ArrayOutput), + }[location] + arguments = '{"text":' + value + "}" + result, _, _ = run( + events(arguments), kwargs=dict(tool=ToolOutput(output, name="answer", strict=True)) + ) + failure(result, "output", known=USAGE, complete=True) + + +@pytest.mark.parametrize("primary", ["success", "whitespace", "timeout"]) +@pytest.mark.parametrize("mask_close_timeout", [False, True]) +def test_stalled_cooperative_finally_close_has_its_own_deadline( + primary: str, mask_close_timeout: bool, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(api(), "STREAM_CLOSE_DEADLINE_SECONDS", 0.005, raising=False) + monkeypatch.setattr(api(), "STREAM_DEADLINE_SECONDS", 0.005 if primary == "timeout" else 60) + + async def scenario() -> Any: + release = asyncio.Event() + + class StalledBody(Body): + attempts = 0 + + async def aclose(self) -> None: + self.attempts += 1 + # A confirmed native close is idempotent; it must not reach the + # second-call stall during the transport's final confirmation. + if primary == "timeout" and self.attempts == 1: + await super().aclose() + return + try: + await release.wait() + except asyncio.CancelledError: + if mask_close_timeout: + raise RuntimeError("sensitive close timeout details") from None + raise + await super().aclose() + + rows = events() if primary != "timeout" else events()[:1] + if primary != "success": + rows[0]["response"]["usage"] = dict(input_tokens=10, output_tokens=5, total_tokens=15) + if primary == "whitespace": + rows[2]["delta"] = " " * 257 + body = StalledBody(rows, wait=primary == "timeout") + # The release keeps the RED run finite without timing assertions. The fixed + # path must finish before this fallback event fires and leave no cleanup task. + timer = asyncio.get_running_loop().call_later(0.03, release.set) + try: + result, _, _ = await harness(body=body) + assert not release.is_set() + assert not isinstance(result, asyncio.CancelledError) + failure( + result, + "timeout" if primary == "success" else primary, + known=USAGE, + complete=primary == "success", + closed=primary == "timeout", + ) + assert body.attempts == 1 + finally: + timer.cancel() + + asyncio.run(scenario()) + + +@pytest.mark.parametrize("primary", ["timeout", "cancelled", "transport"]) +def test_native_finally_cooperative_close_is_bounded_on_pending_read_failure( + primary: str, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(api(), "STREAM_CLOSE_DEADLINE_SECONDS", 0.005) + monkeypatch.setattr(api(), "STREAM_DEADLINE_SECONDS", 0.005 if primary == "timeout" else 60) + + async def scenario() -> None: + release = asyncio.Event() + + class StalledBody(Body): + attempts = 0 + + async def aclose(self) -> None: + self.attempts += 1 + await release.wait() + await super().aclose() + + rows = events()[:1] + rows[0]["response"]["usage"] = dict(input_tokens=10, output_tokens=5, total_tokens=15) + body = StalledBody(rows, wait=primary != "transport", read_error=primary == "transport") + timer = asyncio.get_running_loop().call_later(0.03, release.set) + try: + result, _, _ = await harness(body=body, cancel=primary == "cancelled") + assert not release.is_set() + assert not body.closed + assert body.attempts == 1 + if primary == "cancelled": + assert isinstance(result, api().GuardedResponsesCancelled) + assert result.reason == "cancelled" + assert result.known_usage.model_dump() == USAGE + assert result.usage_complete is False + assert result.stream_closed is False + else: + failure(result, primary, known=USAGE, closed=False) + finally: + timer.cancel() + + asyncio.run(scenario()) + + +def test_native_cancelled_error_survives_close_failure_without_task_cancel_count() -> None: + class CancelledBody(Body): + async def __aiter__(self) -> AsyncIterator[bytes]: + async for chunk in super().__aiter__(): + yield chunk + raise asyncio.CancelledError("sensitive transport cancellation") + + result, _, _ = run(body=CancelledBody(events()[:1], close_error=True)) + assert isinstance(result, api().GuardedResponsesCancelled) + assert result.reason == "cancelled" + assert result.stream_closed is False + assert str(result) == "Guarded Responses request failed" + + +@pytest.mark.parametrize("number, expected", [("1e308", 1e308), ("-1e308", -1e308), ("0.0", 0.0)]) +def test_finite_float_numbers_still_validate(number: str, expected: float) -> None: + class FloatOutput(BaseModel): + model_config = ConfigDict(extra="forbid") + text: float + + result, _, _ = run( + events('{"text":' + number + "}"), + kwargs=dict(tool=ToolOutput(FloatOutput, name="answer", strict=True)), + ) + assert isinstance(result, api().GuardedResponsesToolResult) + assert result.output.text == expected + + +@pytest.mark.parametrize("primary", ["transport", "timeout"]) +@pytest.mark.parametrize("mask_cancellation", [False, True]) +def test_native_cleanup_external_cancel_keeps_earlier_primary_reason( + primary: str, mask_cancellation: bool, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(api(), "STREAM_DEADLINE_SECONDS", 0.005 if primary == "timeout" else 60) + monkeypatch.setattr(api(), "STREAM_CLOSE_DEADLINE_SECONDS", 0.05) + + class ClosingBody(Body): + attempts = 0 + + async def __aiter__(self) -> AsyncIterator[bytes]: + async for chunk in super().__aiter__(): + yield chunk + # The transport mode raises ReadError in super; timeout mode waits for + # the operation timer without signalling the test's close-start event. + await asyncio.Event().wait() + + async def aclose(self) -> None: + self.attempts += 1 + self.blocked.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + if mask_cancellation: + raise RuntimeError("sensitive native cleanup details") from None + raise + + rows = events()[:1] + rows[0]["response"]["usage"] = dict(input_tokens=10, output_tokens=5, total_tokens=15) + body = ClosingBody(rows, read_error=primary == "transport") + result, _, _ = run(body=body, cancel=True) + assert isinstance(result, api().GuardedResponsesCancelled) + assert result.reason == primary + assert result.known_usage.model_dump() == USAGE + assert result.usage_complete is False + assert result.stream_closed is False + assert body.attempts == 1 + assert not body.closed + assert str(result) == "Guarded Responses request failed" + assert result.__cause__ is None