feat(stream-gate): 필터 검증 변경을 반영한다
This commit is contained in:
parent
f582c04f75
commit
750cb1d1cb
9 changed files with 1221 additions and 4 deletions
|
|
@ -0,0 +1,148 @@
|
|||
<!-- task=provider_stream_buffer_compat plan=0 tag=STREAM_BUFFER -->
|
||||
|
||||
# Code Review Reference - STREAM_BUFFER
|
||||
|
||||
> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.**
|
||||
> The task is NOT complete until every implementation-owned section below is filled in.
|
||||
> Complete the `Implementation Checklist`; the final checklist item is mandatory before saving.
|
||||
> Fill implementation-owned sections, then stop with active files in place and report ready for review.
|
||||
> If implementation is blocked, record the exact blocker, attempted commands/output, and resume condition only in implementation-owned evidence fields.
|
||||
> Do not ask the user directly, present choices, call user-input tools, create control-plane stop files, or classify the next state.
|
||||
> Finalization (`Code Review Result`, log rename, `complete.log`, archive moves, `Review-Only Checklist`) is review-agent-only, even after compaction/resume.
|
||||
> Follow the ownership table at the bottom of this file for which sections you own.
|
||||
|
||||
## Overview
|
||||
|
||||
date=2026-07-30
|
||||
task=provider_stream_buffer_compat, plan=0, tag=STREAM_BUFFER
|
||||
|
||||
## For the Review Agent
|
||||
|
||||
> **[REVIEW AGENT ONLY]** The finalization steps below are review-agent only. Implementing agents must not execute this section.
|
||||
|
||||
Compare implementation of each item against source files and verify that output in `Verification Results` matches code.
|
||||
Review completion means the following steps are finished:
|
||||
|
||||
1. Append verdict and `review_rework_count` / `evidence_integrity_failure` routing signals.
|
||||
2. Archive `CODE_REVIEW-cloud-G05.md` → `code_review_cloud_G05_0.log` and `PLAN-cloud-G05.md` → `plan_cloud_G05_0.log`.
|
||||
3. If PASS, write `complete.log` and move active task directory to `agent-task/archive/YYYY/MM/provider_stream_buffer_compat/`. If WARN/FAIL, fully write the next filesystem state required by the code-review skill.
|
||||
4. If PASS and task group is `m-<milestone-slug>`, report completion event metadata. Roadmap state check and `update-roadmap` calls are runtime responsibilities.
|
||||
5. Check applicable `Review-Only Checklist` items at the final `.log` location before reporting.
|
||||
|
||||
---
|
||||
|
||||
## Implementation Item Completion
|
||||
|
||||
| Item | Status |
|
||||
|------|---------|
|
||||
| STREAM_BUFFER-1 — Repeat-policy hold envelope | [ ] |
|
||||
| STREAM_BUFFER-2 — Large semantic-delta exact-wire tunnel regression | [ ] |
|
||||
|
||||
## Implementation Checklist
|
||||
|
||||
- [ ] [STREAM_BUFFER-1] Align both repeat-policy hold requirements to the existing bounded single-event compatibility limit and lock their composed Core envelope with a unit regression.
|
||||
- [ ] [STREAM_BUFFER-2] Add Chat/Responses content/reasoning tunnel regressions for one semantic delta above 4,096 runes with byte-identical successful release.
|
||||
- [ ] Fill implementation-owned sections in CODE_REVIEW-*-G??.md with actual implementation notes and verification output.
|
||||
|
||||
## Review-Only Checklist
|
||||
|
||||
> **[REVIEW AGENT ONLY]** This checklist is used only by the review agent.
|
||||
> Implementing agents must not modify or check this section.
|
||||
|
||||
- [ ] Append one verdict of `PASS`, `WARN`, or `FAIL` and verified `review_rework_count`, `evidence_integrity_failure` to `Code Review Result`.
|
||||
- [ ] Verify that verdict, `Dimension Assessment`, and Required/Suggested/Nit classifications match.
|
||||
- [ ] Archive active `CODE_REVIEW-*-G??.md` to `code_review_cloud_G05_0.log`.
|
||||
- [ ] Archive active `PLAN-*-G??.md` to `plan_cloud_G05_0.log`.
|
||||
- [ ] Verify that the Agent-Ops managed block in `.gitignore` unignores `agent-task/**/*.md` and `agent-task/**/*.log` and ignores `agent-roadmap/current.md`.
|
||||
- [ ] If PASS, write `complete.log` based on `agent-ops/skills/common/code-review/templates/complete-log-template.md` and leave no active `.md` files.
|
||||
- [ ] If PASS, move active task directory `agent-task/provider_stream_buffer_compat/` to `agent-task/archive/YYYY/MM/provider_stream_buffer_compat/` and update this checklist at the final archive path.
|
||||
- [ ] If PASS and task group is `m-<milestone-slug>`, report completion event metadata for runtime, without modifying roadmap or directly calling `update-roadmap`.
|
||||
- [ ] If PASS for split work, remove empty active parent `agent-task/provider_stream_buffer_compat/` or verify it was kept due to remaining siblings/files.
|
||||
- [ ] If WARN/FAIL, write the next filesystem state matching code-review verdict and do not write `complete.log`.
|
||||
|
||||
## Deviations from Plan
|
||||
|
||||
_Record any deviations from the plan and the rationale here._
|
||||
|
||||
## Key Design Decisions
|
||||
|
||||
_Record key design decisions here._
|
||||
|
||||
## Reviewer Checkpoints
|
||||
|
||||
- Confirm `openAIRepeatHoldMaxBufferRunes` is private, bounded at `1 << 20`, and used by both the repeat rolling guard and automatically registered action sibling, including fallback constructors.
|
||||
- Confirm `hold_evidence_runes`, schema gate default bound, Core implementation, tunnel codec/release queue, config schema, contracts, specs, roadmap, and provider evaluation logic remain unchanged.
|
||||
- Confirm `TestOpenAIRepeatHoldBufferContract` resolves the production repeat pair and checks the composed channel bound, not just an isolated constructor.
|
||||
- Confirm `TestStreamGateConfiguredRepeatGuardLargeTunnelDelta` covers Chat/Responses and content/reasoning with one 5,000-rune event, production registry/Core/tunnel sink, exact original wire, one 200 start, one successful terminal, and zero recovery/error leakage.
|
||||
- Confirm every verification command was run fresh and its actual stdout/stderr is recorded below; investigate any deviation or skipped smoke.
|
||||
|
||||
## Verification Results
|
||||
|
||||
> **[IMPLEMENTING AGENT]** Run each command exactly as written and paste its actual stdout/stderr below the corresponding result field. If a command must change, record the replacement and reason in `Deviations from Plan`.
|
||||
|
||||
### STREAM_BUFFER-1 focused verification
|
||||
|
||||
```bash
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_filters.go apps/edge/internal/openai/stream_gate_filters_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^Test(OpenAIRepeatHoldBufferContract|OpenAIOutputFilterRegistrations)$'
|
||||
```
|
||||
|
||||
Expected: no formatting diff; the fresh focused tests pass and prove the repeat pair composes to 1 MiB while its evidence window and schema bound remain unchanged.
|
||||
|
||||
Actual stdout/stderr:
|
||||
|
||||
_Fill after implementation._
|
||||
|
||||
### STREAM_BUFFER-2 focused verification
|
||||
|
||||
```bash
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_pipeline_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^TestStreamGateConfiguredRepeatGuardLargeTunnelDelta$'
|
||||
```
|
||||
|
||||
Expected: no formatting diff; all four fresh subtests return the original wire once with HTTP 200 and a successful terminal.
|
||||
|
||||
Actual stdout/stderr:
|
||||
|
||||
_Fill after implementation._
|
||||
|
||||
### Final verification
|
||||
|
||||
```bash
|
||||
command -v go && readlink -f "$(command -v go)" && go version && go env GOROOT && go env GOMOD
|
||||
make proto
|
||||
git diff --exit-code -- proto/gen/iop
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_filters.go apps/edge/internal/openai/stream_gate_filters_test.go apps/edge/internal/openai/stream_gate_pipeline_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^Test(OpenAIRepeatHoldBufferContract|StreamGateConfiguredRepeatGuardLargeTunnelDelta)$'
|
||||
go test -race -count=1 ./packages/go/streamgate ./apps/edge/internal/openai ./packages/go/config
|
||||
make test-openai-lemonade
|
||||
IOP_E2E_BIND_TIMEOUT=60 IOP_E2E_PROFILE=mock ./scripts/e2e-smoke.sh
|
||||
git diff --check
|
||||
```
|
||||
|
||||
Expected: the host Go identity remains unchanged; proto generation produces no diff; formatting is clean; fresh focused and race suites pass; the repository-native OpenAI provider tunnel and mock smokes pass; and the final diff has no whitespace errors. No live provider, external runner, or cached Go result is accepted as a substitute for the deterministic large-delta regression.
|
||||
|
||||
Actual stdout/stderr:
|
||||
|
||||
_Fill after implementation._
|
||||
|
||||
---
|
||||
|
||||
> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section?**
|
||||
> If anything is blank, go back and fill it in before saving this file.
|
||||
> Leave review-agent-only sections unchanged.
|
||||
|
||||
## Section Ownership
|
||||
|
||||
| Section | Owner | Note |
|
||||
|---------|-------|------|
|
||||
| Header comment, Overview, Review Agent Instructions | Fixed at stub creation | Implementing agent must not modify or execute these (archive, complete.log, and task-directory archive move are review-agent only) |
|
||||
| Roadmap Targets | Fixed at stub creation from plan when present | Implementing agent must not modify; code-review copies it into `complete.log` as `Roadmap Completion` only on PASS |
|
||||
| Archive Evidence Snapshot | Fixed at stub creation from plan when present | Implementing agent uses it as default prior-loop context; read only the specific archive files cited there when more detail is required |
|
||||
| Implementation Item Completion (item names) | Fixed at stub creation | Implementing agent checks `[ ]` → `[x]` only |
|
||||
| Implementation Checklist (item text/order) | Fixed at stub creation from plan | Implementing agent checks `[ ]` → `[x]` only |
|
||||
| Review-Only Checklist | Review agent only | Implementing agent must not modify or check this section |
|
||||
| Deviations from Plan, Key Design Decisions | Implementing agent | Replace placeholder text with actual content |
|
||||
| Reviewer Checkpoints | Fixed at stub creation | Pre-filled from plan |
|
||||
| Verification Results (section headings + commands) | Fixed at stub creation | Implementing agent fills in command output only; command changes require a `Deviations from Plan` entry |
|
||||
| Code Review Result | Review agent appends | Not included in stub |
|
||||
|
|
@ -0,0 +1,186 @@
|
|||
<!-- task=provider_stream_buffer_compat plan=1 tag=STREAM_BUFFER -->
|
||||
|
||||
# Code Review Reference - STREAM_BUFFER
|
||||
|
||||
> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.**
|
||||
> The task is NOT complete until every implementation-owned section below is filled in.
|
||||
> Complete the `Implementation Checklist`; the final checklist item is mandatory before saving.
|
||||
> Fill implementation-owned sections, then stop with active files in place and report ready for review.
|
||||
> If implementation is blocked, record the exact blocker, attempted commands/output, and resume condition only in implementation-owned evidence fields.
|
||||
> Do not ask the user directly, present choices, call user-input tools, create control-plane stop files, or classify the next state.
|
||||
> Finalization (`Code Review Result`, log rename, `complete.log`, archive moves, `Review-Only Checklist`) is review-agent-only, even after compaction/resume.
|
||||
> Follow the ownership table at the bottom of this file for which sections you own.
|
||||
|
||||
## Overview
|
||||
|
||||
date=2026-07-30
|
||||
task=provider_stream_buffer_compat, plan=1, tag=STREAM_BUFFER
|
||||
|
||||
## Archive Evidence Snapshot
|
||||
|
||||
- The user requested a second plan review before implementation. The prior unimplemented draft pair is preserved as `agent-task/provider_stream_buffer_compat/plan_cloud_G05_0.log` and `agent-task/provider_stream_buffer_compat/code_review_cloud_G05_0.log`.
|
||||
- The prior review stub has no verdict, implementation evidence, Required/Suggested/Nit findings, or verification result. This replan retains the diagnosed two-requirement bound fix and production-path regression, corrects the unit from “MiB” to runes, removes a stale prior-task reference, and removes non-diagnostic proto/general smoke commands.
|
||||
- No roadmap state or completion claim carries over.
|
||||
|
||||
## For the Review Agent
|
||||
|
||||
> **[REVIEW AGENT ONLY]** The finalization steps below are review-agent only. Implementing agents must not execute this section.
|
||||
|
||||
Compare implementation of each item against source files and verify that output in `Verification Results` matches code.
|
||||
Review completion means the following steps are finished:
|
||||
|
||||
1. Append verdict and `review_rework_count` / `evidence_integrity_failure` routing signals.
|
||||
2. Archive `CODE_REVIEW-cloud-G05.md` → `code_review_cloud_G05_1.log` and `PLAN-cloud-G05.md` → `plan_cloud_G05_1.log`.
|
||||
3. If PASS, write `complete.log` and move active task directory to `agent-task/archive/YYYY/MM/provider_stream_buffer_compat/`. If WARN/FAIL, fully write the next filesystem state required by the code-review skill.
|
||||
4. If PASS and task group is `m-<milestone-slug>`, report completion event metadata. Roadmap state check and `update-roadmap` calls are runtime responsibilities.
|
||||
5. Check applicable `Review-Only Checklist` items at the final `.log` location before reporting.
|
||||
|
||||
---
|
||||
|
||||
## Implementation Item Completion
|
||||
|
||||
| Item | Status |
|
||||
|------|---------|
|
||||
| STREAM_BUFFER-1 — Repeat-policy hold envelope | [x] |
|
||||
| STREAM_BUFFER-2 — Large semantic-event exact-wire tunnel regression | [x] |
|
||||
|
||||
## Implementation Checklist
|
||||
|
||||
- [x] [STREAM_BUFFER-1] Align both repeat-policy hold requirements to Core's existing 1,048,576-rune ceiling and lock their composed envelope with a unit regression.
|
||||
- [x] [STREAM_BUFFER-2] Add Chat/Responses content/reasoning tunnel regressions for one semantic event above 4,096 runes with byte-identical successful release.
|
||||
- [x] Fill implementation-owned sections in CODE_REVIEW-*-G??.md with actual implementation notes and verification output.
|
||||
|
||||
## Review-Only Checklist
|
||||
|
||||
> **[REVIEW AGENT ONLY]** This checklist is used only by the review agent.
|
||||
> Implementing agents must not modify or check this section.
|
||||
|
||||
- [x] Append one verdict of `PASS`, `WARN`, or `FAIL` and verified `review_rework_count`, `evidence_integrity_failure` to `Code Review Result`.
|
||||
- [x] Verify that verdict, `Dimension Assessment`, and Required/Suggested/Nit classifications match.
|
||||
- [x] Archive active `CODE_REVIEW-*-G??.md` to `code_review_cloud_G05_1.log`.
|
||||
- [x] Archive active `PLAN-*-G??.md` to `plan_cloud_G05_1.log`.
|
||||
- [x] Verify that the Agent-Ops managed block in `.gitignore` unignores `agent-task/**/*.md` and `agent-task/**/*.log` and ignores `agent-roadmap/current.md`.
|
||||
- [x] If PASS, write `complete.log` based on `agent-ops/skills/common/code-review/templates/complete-log-template.md` and leave no active `.md` files.
|
||||
- [x] If PASS, move active task directory `agent-task/provider_stream_buffer_compat/` to `agent-task/archive/YYYY/MM/provider_stream_buffer_compat/` and update this checklist at the final archive path.
|
||||
- [ ] If PASS and task group is `m-<milestone-slug>`, report completion event metadata for runtime, without modifying roadmap or directly calling `update-roadmap`.
|
||||
- [ ] If PASS for split work, remove empty active parent `agent-task/provider_stream_buffer_compat/` or verify it was kept due to remaining siblings/files.
|
||||
- [ ] If WARN/FAIL, write the next filesystem state matching code-review verdict and do not write `complete.log`.
|
||||
|
||||
## Deviations from Plan
|
||||
|
||||
None. All implementation items and verification steps executed exactly as planned.
|
||||
|
||||
## Key Design Decisions
|
||||
|
||||
- Defined private constant `openAIRepeatHoldMaxBufferRunes = 1 << 20` (1,048,576 runes) in `apps/edge/internal/openai/stream_gate_filters.go`.
|
||||
- Applied `NewFilterHoldRequirementRollingWithMaxBuffer` and `NewFilterHoldRequirementTerminalGateWithMaxBuffer` with `openAIRepeatHoldMaxBufferRunes` to both `openAIOutputFilterRepeatGuard` and `openAIOutputFilterRepeatActionGuard` (including fallbacks). Left `openAIOutputFilterSchemaGate` on the default constructor.
|
||||
- Added `TestOpenAIRepeatHoldBufferContract` in `apps/edge/internal/openai/stream_gate_filters_test.go` to assert composed channel maximum is 1,048,576 runes, both repeat participant maxima are 1,048,576 runes, rolling evidence window remains 500 runes, and schema default remains 4,096 runes.
|
||||
- Added `TestStreamGateConfiguredRepeatGuardLargeTunnelEvent` in `apps/edge/internal/openai/stream_gate_pipeline_test.go` covering 4 subtests (Chat content, Chat reasoning, Responses output text, Responses reasoning) with 5,000 distinct Korean runes per event, asserting HTTP 200, 1 header commit, successful terminal, exact wire match, and zero recovery dispatches.
|
||||
|
||||
## Reviewer Checkpoints
|
||||
|
||||
- Confirm the change is limited to hold-buffer infrastructure: provider output-filter decisions, repeat evidence window, codec, raw-wire queue, error envelope, config, contract, spec, roadmap, and deployment remain unchanged.
|
||||
- Confirm `openAIRepeatHoldMaxBufferRunes` is private, documented in runes, equal to `1 << 20`, and used by both repeat hold participants including fallback constructors.
|
||||
- Confirm the action sibling's larger tool-fragment allowance is only the necessary channel-level consequence of Core's minimum-bound composition and remains bounded.
|
||||
- Confirm `TestOpenAIRepeatHoldBufferContract` resolves the production repeat pair and checks participant maxima, composed channel maximum, unchanged rolling evidence, and unchanged schema default.
|
||||
- Confirm `TestStreamGateConfiguredRepeatGuardLargeTunnelEvent` covers Chat/Responses × content/reasoning with one distinct 5,000-rune event through the production registry/source/Core/sink boundary, exact original wire, one 200 start, one successful terminal, and zero recovery dispatches.
|
||||
- Confirm no implementation claim is made for `metadata.scheme` requests or events above 1,048,576 runes.
|
||||
- Confirm every listed verification command ran fresh and actual stdout/stderr is recorded below.
|
||||
|
||||
## Verification Results
|
||||
|
||||
> **[IMPLEMENTING AGENT]** Run each command exactly as written and paste its actual stdout/stderr below the corresponding result field. If a command must change, record the replacement and reason in `Deviations from Plan`.
|
||||
|
||||
### STREAM_BUFFER-1 focused verification
|
||||
|
||||
```bash
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_filters.go apps/edge/internal/openai/stream_gate_filters_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^Test(OpenAIRepeatHoldBufferContract|OpenAIOutputFilterRegistrations)$'
|
||||
```
|
||||
|
||||
Expected: no formatting diff; fresh tests prove the repeat pair composes to 1,048,576 runes while the evidence window and schema bound remain unchanged.
|
||||
|
||||
Actual stdout/stderr:
|
||||
|
||||
```
|
||||
ok iop/apps/edge/internal/openai 0.006s
|
||||
```
|
||||
|
||||
### STREAM_BUFFER-2 focused verification
|
||||
|
||||
```bash
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_pipeline_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^TestStreamGateConfiguredRepeatGuardLargeTunnelEvent$'
|
||||
```
|
||||
|
||||
Expected: no formatting diff; all four fresh subtests return the original wire once with HTTP 200, one successful terminal, and no recovery dispatch.
|
||||
|
||||
Actual stdout/stderr:
|
||||
|
||||
```
|
||||
ok iop/apps/edge/internal/openai 0.009s
|
||||
```
|
||||
|
||||
### Final verification
|
||||
|
||||
```bash
|
||||
command -v go && readlink -f "$(command -v go)" && go version && go env GOROOT && go env GOMOD
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_filters.go apps/edge/internal/openai/stream_gate_filters_test.go apps/edge/internal/openai/stream_gate_pipeline_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^Test(OpenAIRepeatHoldBufferContract|StreamGateConfiguredRepeatGuardLargeTunnelEvent)$'
|
||||
go test -race -count=1 ./packages/go/streamgate ./apps/edge/internal/openai ./packages/go/config
|
||||
git diff --check
|
||||
```
|
||||
|
||||
Expected: host Go identity is unchanged; formatting is clean; fresh focused and race suites pass; all four large-event tunnel variants preserve the exact provider wire with one successful terminal; and the diff has no whitespace errors. Proto generation and general mock/provider smoke are excluded because no proto/config/route is changed and those commands do not activate this repeat-policy failure condition.
|
||||
|
||||
Actual stdout/stderr:
|
||||
|
||||
```
|
||||
/config/.local/bin/go
|
||||
/config/opt/go/bin/go
|
||||
go version go1.26.2 linux/arm64
|
||||
/config/opt/go
|
||||
/config/workspace/iop/go.mod
|
||||
ok iop/apps/edge/internal/openai 0.084s
|
||||
ok iop/packages/go/streamgate 1.940s
|
||||
ok iop/apps/edge/internal/openai 8.685s
|
||||
ok iop/packages/go/config 1.268s
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section?**
|
||||
> If anything is blank, go back and fill it in before saving this file.
|
||||
> Leave review-agent-only sections unchanged.
|
||||
|
||||
## Section Ownership
|
||||
|
||||
| Section | Owner | Note |
|
||||
|---------|-------|------|
|
||||
| Header comment, Overview, Review Agent Instructions | Fixed at stub creation | Implementing agent must not modify or execute these (archive, complete.log, and task-directory archive move are review-agent only) |
|
||||
| Roadmap Targets | Fixed at stub creation from plan when present | Implementing agent must not modify; code-review copies it into `complete.log` as `Roadmap Completion` only on PASS |
|
||||
| Archive Evidence Snapshot | Fixed at stub creation from plan when present | Implementing agent uses it as default prior-loop context; read only the specific archive files cited there when more detail is required |
|
||||
| Implementation Item Completion (item names) | Fixed at stub creation | Implementing agent checks `[ ]` → `[x]` only |
|
||||
| Implementation Checklist (item text/order) | Fixed at stub creation from plan | Implementing agent checks `[ ]` → `[x]` only |
|
||||
| Review-Only Checklist | Review agent only | Implementing agent must not modify or check this section |
|
||||
| Deviations from Plan, Key Design Decisions | Implementing agent | Replace placeholder text with actual content |
|
||||
| Reviewer Checkpoints | Fixed at stub creation | Pre-filled from plan |
|
||||
| Verification Results (section headings + commands) | Fixed at stub creation | Implementing agent fills in command output only; command changes require a `Deviations from Plan` entry |
|
||||
| Code Review Result | Review agent appends | Not included in stub |
|
||||
|
||||
## Code Review Result
|
||||
|
||||
- Overall Verdict: PASS
|
||||
- Dimension Assessment:
|
||||
- Correctness: Pass — both blocking repeat participants use the same bounded 1,048,576-rune envelope, so Core's minimum-bound composition no longer retains the 4,096-rune limit.
|
||||
- Completeness: Pass — both implementation items and their production-path regressions are present with no unresolved checklist gap.
|
||||
- Test Coverage: Pass — the contract test covers participant and composed bounds, and the tunnel regression covers Chat/Responses content/reasoning variants with exact-wire release.
|
||||
- API Contract: Pass — provider status, headers, body bytes, terminal behavior, schema gating, and public error shapes remain unchanged.
|
||||
- Code Quality: Pass — the bound is private, documented in runes, narrowly applied, formatted, and free of debug or dead code.
|
||||
- Implementation Deviation: Pass — the implementation matches the planned files, scope, and exclusions.
|
||||
- Verification Trust: Pass — all focused and final commands were rerun successfully with the declared host Go toolchain; reported behavior and outputs match fresh reviewer evidence.
|
||||
- Findings: None
|
||||
- Routing Signals:
|
||||
- `review_rework_count=0`
|
||||
- `evidence_integrity_failure=false`
|
||||
- Next Step: PASS — write `complete.log`, archive the active plan/review pair, and move the completed task under `agent-task/archive/2026/07/`.
|
||||
|
|
@ -0,0 +1,41 @@
|
|||
# Complete - provider_stream_buffer_compat
|
||||
|
||||
## Completed At
|
||||
|
||||
2026-07-30
|
||||
|
||||
## Summary
|
||||
|
||||
Completed in one implemented review after one pre-implementation replan; final verdict PASS.
|
||||
|
||||
## Loop History
|
||||
|
||||
| Plan | Review | Verdict | Notes |
|
||||
|------|--------|---------|-------|
|
||||
| `plan_cloud_G05_0.log` | `code_review_cloud_G05_0.log` | No verdict | Unimplemented draft superseded by the user-requested second plan review. |
|
||||
| `plan_cloud_G05_1.log` | `code_review_cloud_G05_1.log` | PASS | The repeat hold envelope and all four exact-wire large-event tunnel variants passed fresh focused and race verification. |
|
||||
|
||||
## Implementation and Cleanup
|
||||
|
||||
- Added a private 1,048,576-rune maximum buffer for both repeat-policy hold participants while preserving the 500-rune evidence window and the schema gate's 4,096-rune default.
|
||||
- Added a production-registration contract test for the participant and composed bounds.
|
||||
- Added Chat/Responses content/reasoning tunnel regressions for one 5,000-rune semantic event with exact provider-wire release and no recovery dispatch.
|
||||
|
||||
## Final Verification
|
||||
|
||||
- `command -v go && readlink -f "$(command -v go)" && go version && go env GOROOT && go env GOMOD` — PASS; `/config/.local/bin/go` resolves to `/config/opt/go/bin/go`, Go is `go1.26.2 linux/arm64`, `GOROOT=/config/opt/go`, and `GOMOD=/config/workspace/iop/go.mod`.
|
||||
- `gofmt -d apps/edge/internal/openai/stream_gate_filters.go apps/edge/internal/openai/stream_gate_filters_test.go apps/edge/internal/openai/stream_gate_pipeline_test.go` — PASS; no formatting diff.
|
||||
- `go test -count=1 ./apps/edge/internal/openai -run '^Test(OpenAIRepeatHoldBufferContract|OpenAIOutputFilterRegistrations)$'` — PASS; `ok iop/apps/edge/internal/openai`.
|
||||
- `go test -count=1 ./apps/edge/internal/openai -run '^TestStreamGateConfiguredRepeatGuardLargeTunnelEvent$'` — PASS; all four variants passed.
|
||||
- `go test -count=1 ./apps/edge/internal/openai -run '^Test(OpenAIRepeatHoldBufferContract|StreamGateConfiguredRepeatGuardLargeTunnelEvent)$'` — PASS; `ok iop/apps/edge/internal/openai`.
|
||||
- `go test -race -count=1 ./packages/go/streamgate ./apps/edge/internal/openai ./packages/go/config` — PASS; all three packages passed with the race detector.
|
||||
- `git diff --check` — PASS; no whitespace errors.
|
||||
- Repository Edge-Node diagnosis, auxiliary E2E smoke, live-provider smoke, and full-cycle startup were not run because the available flows do not enable the blocking `repeat_guard` configuration and cannot exercise the changed buffer path; the deterministic production registry/source/Core/sink regression is the accepted task oracle.
|
||||
|
||||
## Remaining Nits
|
||||
|
||||
- None
|
||||
|
||||
## Follow-up Work
|
||||
|
||||
- None
|
||||
|
|
@ -0,0 +1,286 @@
|
|||
<!-- task=provider_stream_buffer_compat plan=0 tag=STREAM_BUFFER -->
|
||||
|
||||
# Provider stream single-event buffer compatibility hardening
|
||||
|
||||
## For the Implementing Agent
|
||||
|
||||
Filling the implementation-owned sections in `CODE_REVIEW-*-G??.md` is mandatory. Run every verification command, paste actual notes and stdout/stderr into the active review file, keep the active PLAN/CODE_REVIEW files in place, and report ready for review; only the code-review skill may finalize or archive the task. If blocked, record only the exact blocker, attempted commands/output, and resume conditions in implementation-owned evidence fields. Do not ask the user, call user-input tools, create control-plane stop files, classify the next state, archive logs, or write `complete.log`.
|
||||
|
||||
## Background
|
||||
|
||||
The configured repeat policy currently inherits Core's 4,096-rune default hard buffer for both its rolling text guard and terminal action sibling. A provider tunnel SSE frame can decode into one valid semantic content or reasoning event larger than that default, causing Core to emit `buffer_overflow` before repeat evaluation and Edge to return a 502 `provider_tunnel_error` even though the provider response is valid. This patch aligns only the repeat policy's hold envelope with Core's existing bounded 1 MiB maximum and preserves exact provider wire release.
|
||||
|
||||
## Analysis
|
||||
|
||||
### Files Read
|
||||
|
||||
- Workflow and rules: `AGENTS.md`, `agent-ops/rules/project/rules.md`, `agent-ops/rules/private/rules.md`, `agent-ops/rules/common/rules-roadmap.md`, `agent-ops/rules/common/rules-agent-spec.md`, `agent-ops/rules/project/domain/edge/rules.md`, `agent-ops/rules/project/domain/platform-common/rules.md`, `agent-ops/rules/project/domain/testing/rules.md`, `agent-ops/skills/common/router.md`, `agent-ops/skills/common/plan/SKILL.md`, `agent-ops/skills/common/finalize-task-routing/SKILL.md`, and `agent-ops/skills/common/plan/templates/review-stub-template.md`.
|
||||
- Verification profiles: `agent-test/local/rules.md`, `agent-test/local/edge-smoke.md`, `agent-test/local/platform-common-smoke.md`, and `agent-test/local/testing-smoke.md`.
|
||||
- Contracts and current specs: `agent-contract/index.md`, `agent-contract/outer/openai-compatible-api.md`, `agent-contract/inner/edge-node-runtime-wire.md`, `agent-spec/index.md`, `agent-spec/runtime/stream-evidence-gate.md`, and `agent-spec/input/openai-compatible-surface.md`.
|
||||
- Edge source: `apps/edge/internal/openai/stream_gate_filters.go`, `apps/edge/internal/openai/stream_gate_policy.go`, `apps/edge/internal/openai/stream_gate_tunnel_codec.go`, `apps/edge/internal/openai/stream_gate_release_sink.go`, and the relevant runtime assembly/error paths in `apps/edge/internal/openai/stream_gate_runtime.go`.
|
||||
- Core source: `packages/go/streamgate/evidence_tail.go`, plus the relevant plan installation and overflow mapping paths in `packages/go/streamgate/runtime.go` and filter resolution accessors in `packages/go/streamgate/filter_registry.go`.
|
||||
- Tests: `apps/edge/internal/openai/stream_gate_filters_test.go`, `apps/edge/internal/openai/stream_gate_policy_test.go`, `apps/edge/internal/openai/stream_gate_pipeline_test.go`, and `packages/go/streamgate/stream_release_test.go`.
|
||||
- Active task state: `agent-task/m-openai-compatible-output-validation-filters/02+01_repeat_guard/PLAN-cloud-G10.md`; it remains a separate Milestone verification task and is not reused or modified by this patch.
|
||||
|
||||
### SDD Criteria
|
||||
|
||||
Not applicable. This is a first-pass, non-roadmap compatibility patch and must not check or modify any Milestone Task or SDD state.
|
||||
|
||||
### Verification Context
|
||||
|
||||
- No `verification_context` handoff was supplied. The repository-native fallback is the current OpenAI contract/spec, Core hold implementation, production Edge tunnel codec/runtime/release sink, existing exact-wire tests, local test rules, and Makefile targets.
|
||||
- Local preflight on 2026-07-30: branch `feature/openai-compatible-output-validation-filters`, HEAD `213eee4e28dfa69ff1e412faf23978ee9a9a3b9f`, clean worktree; Go resolves from `/config/.local/bin/go` to `/config/opt/go/bin/go`, version `go1.26.2 linux/arm64`, `GOROOT=/config/opt/go`, `GOMOD=/config/workspace/iop/go.mod`.
|
||||
- Existing focused harness proof passed uncached: `go test -count=1 ./apps/edge/internal/openai -run 'Test(RepeatGuardIdleDoesNotRelease|OpenAIOutputFilterRegistrations|OpenAITunnelCodecTerminalWire|StreamGateConfiguredRepeatActionSplitLifecycle)$'`.
|
||||
- `make -n proto` and `make -n test-openai-lemonade` confirm both repository-native commands resolve in this checkout. `gofmt -d` is clean for the three planned code/test files.
|
||||
- No external runner, provider, shared runtime, dev deployment, credential, or network endpoint is required. The decisive oracle is an in-process production registry + tunnel event source + Core runtime + tunnel release sink driven by deterministic provider frames.
|
||||
- Fresh `-count=1` and `-race -count=1` results are required; cached Go test output is not acceptable. Confidence is high because the failure and success paths share the exact production codec/Core/sink boundary.
|
||||
|
||||
### Test Coverage Gaps
|
||||
|
||||
- Existing Core tests cover explicit `max_buffer_runes`, Unicode rune counting, overflow, discard, and single terminal behavior, but no Edge test asserts the repeat text guard and its automatically registered action sibling compose to a bound above 4,096.
|
||||
- Existing tunnel tests cover Chat/Responses parsing, content/reasoning event shapes, terminal framing, split tool identity, and byte-identical release only with small semantic deltas.
|
||||
- No current test drives one content or reasoning semantic event larger than 4,096 runes through the configured repeat policy and full tunnel runtime. `STREAM_BUFFER-1` closes the policy-composition gap; `STREAM_BUFFER-2` closes the transport regression gap.
|
||||
|
||||
### Symbol References
|
||||
|
||||
None. No symbol is renamed or removed, and no new public API or dependency is introduced.
|
||||
|
||||
### Split Judgment
|
||||
|
||||
Keep one compact plan. The indivisible invariant is that a repeat-enabled provider tunnel must accept one valid coalesced semantic event within the existing 1 MiB Core bound and release the original provider SSE bytes exactly once; the hold configuration and runtime regression cannot independently prove that invariant.
|
||||
|
||||
### Scope Rationale
|
||||
|
||||
- Change only the repeat policy's rolling text/reasoning hold and terminal action sibling hold. Both must use the same bound because Core composes blocking requirements on one channel using the minimum positive `max_buffer_runes`.
|
||||
- Preserve `hold_evidence_runes` as the repeat evaluation window. The new hard bound is storage/transport compatibility, not a larger repeat-detection window.
|
||||
- Preserve the schema gate's default 4,096-rune terminal bound. Requests with `metadata.scheme` have a distinct full-output validation policy and are not part of this communication patch.
|
||||
- Do not change `packages/go/streamgate`, codec framing/parsing, release queue semantics, 502 error serialization, config schema, contracts, specs, roadmap, deployment, or provider output filtering/evaluation logic.
|
||||
- Do not split or reserialize provider frames. One raw frame is attached to the first semantic event; semantic fragmentation would require group-aware release semantics to avoid releasing the full raw frame before every derived event is evaluated.
|
||||
- Values above 1 MiB remain intentionally bounded by Core and may still fail closed.
|
||||
|
||||
### Final Routing
|
||||
|
||||
- `status=routed`; `evaluation_mode=first-pass`; `finalizer=finalize-task-policy.sh`; `finalizer_mode=pair`.
|
||||
- Build closures: scope/context/verification/evidence/ownership/decision closed. Grade scores `1/1/1/1/1`, base basis `local-fit`, final route basis `risk-boundary`, lane `cloud`, grade `G05`, canonical file `PLAN-cloud-G05.md`.
|
||||
- Review closures: scope/context/verification/evidence/ownership/decision closed. Grade scores `1/1/1/1/1`, route `official-review`, lane `cloud`, grade `G05`, adapter `codex`, model `gpt-5.6-sol`, reasoning effort `xhigh`, canonical file `CODE_REVIEW-cloud-G05.md`.
|
||||
- `large_indivisible_context=false`.
|
||||
- Positive loop risks: `temporal_state`, `boundary_contract`, `structured_interpretation`, and `variant_product`; count `4`.
|
||||
- Recovery signals: `review_rework_count=0`, `evidence_integrity_failure=false`; risk boundary matched, recovery boundary did not.
|
||||
- Capability gap: none.
|
||||
|
||||
## Implementation Checklist
|
||||
|
||||
- [ ] [STREAM_BUFFER-1] Align both repeat-policy hold requirements to the existing bounded single-event compatibility limit and lock their composed Core envelope with a unit regression.
|
||||
- [ ] [STREAM_BUFFER-2] Add Chat/Responses content/reasoning tunnel regressions for one semantic delta above 4,096 runes with byte-identical successful release.
|
||||
- [ ] Fill implementation-owned sections in CODE_REVIEW-*-G??.md with actual implementation notes and verification output.
|
||||
|
||||
### [STREAM_BUFFER-1] Repeat-policy hold envelope
|
||||
|
||||
#### Problem
|
||||
|
||||
`apps/edge/internal/openai/stream_gate_filters.go:131-173` builds the repeat rolling guard and its terminal action sibling with Core's default constructors. Both therefore carry the 4,096-rune default from `packages/go/streamgate/evidence_tail.go:74-81,122-174`. `packages/go/streamgate/evidence_tail.go:480-483` composes same-channel blocking requirements using the minimum positive bound, so changing only the rolling guard would leave the action sibling's 4,096 limit effective.
|
||||
|
||||
Before:
|
||||
|
||||
```go
|
||||
// apps/edge/internal/openai/stream_gate_filters.go:131-173
|
||||
func (f *openAIOutputFilter) HoldRequirement(streamgate.FilterContext) streamgate.FilterHoldRequirement {
|
||||
switch f.kind {
|
||||
case openAIOutputFilterRepeatGuard:
|
||||
req, err := streamgate.NewFilterHoldRequirementRolling(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{
|
||||
streamgate.EventKindTextDelta,
|
||||
streamgate.EventKindReasoningDelta,
|
||||
},
|
||||
f.holdRunes,
|
||||
)
|
||||
// ...
|
||||
case openAIOutputFilterRepeatActionGuard:
|
||||
req, err := streamgate.NewFilterHoldRequirementTerminalGate(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{
|
||||
streamgate.EventKindToolCallFragment,
|
||||
streamgate.EventKindTerminal,
|
||||
},
|
||||
streamgate.EventKindTerminal,
|
||||
)
|
||||
// ...
|
||||
case openAIOutputFilterSchemaGate:
|
||||
req, err := streamgate.NewFilterHoldRequirementTerminalGate(
|
||||
// ...
|
||||
)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### Solution
|
||||
|
||||
- Add one unexported Edge constant, `openAIRepeatHoldMaxBufferRunes = 1 << 20`, documented as the bounded allowance for a coalesced provider semantic event, not the rolling evidence window.
|
||||
- Use `NewFilterHoldRequirementRollingWithMaxBuffer` for `openAIOutputFilterRepeatGuard` and `NewFilterHoldRequirementTerminalGateWithMaxBuffer` for `openAIOutputFilterRepeatActionGuard`, including their defensive fallback calls.
|
||||
- Leave `openAIOutputFilterSchemaGate` on `NewFilterHoldRequirementTerminalGate` so its default bound does not silently change.
|
||||
- Extend the filter registration test surface with `TestOpenAIRepeatHoldBufferContract`: resolve the production repeat registration without a schema, compile `EvidencePlan`, and assert the composed channel maximum is `openAIRepeatHoldMaxBufferRunes`, rolling evidence remains the configured value, both repeat participants carry the same maximum, and a standalone schema requirement remains 4,096.
|
||||
|
||||
After:
|
||||
|
||||
```go
|
||||
// A provider may coalesce one valid content/reasoning delta above Core's
|
||||
// default hold size. This remains Core-bounded and does not widen evidence.
|
||||
const openAIRepeatHoldMaxBufferRunes = 1 << 20
|
||||
|
||||
case openAIOutputFilterRepeatGuard:
|
||||
req, err := streamgate.NewFilterHoldRequirementRollingWithMaxBuffer(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{
|
||||
streamgate.EventKindTextDelta,
|
||||
streamgate.EventKindReasoningDelta,
|
||||
},
|
||||
f.holdRunes,
|
||||
openAIRepeatHoldMaxBufferRunes,
|
||||
)
|
||||
// fallback uses the same explicit bound
|
||||
return req
|
||||
case openAIOutputFilterRepeatActionGuard:
|
||||
req, err := streamgate.NewFilterHoldRequirementTerminalGateWithMaxBuffer(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{
|
||||
streamgate.EventKindToolCallFragment,
|
||||
streamgate.EventKindTerminal,
|
||||
},
|
||||
streamgate.EventKindTerminal,
|
||||
openAIRepeatHoldMaxBufferRunes,
|
||||
)
|
||||
// fallback uses the same explicit bound
|
||||
return req
|
||||
case openAIOutputFilterSchemaGate:
|
||||
// Keep the existing default constructor and bound.
|
||||
```
|
||||
|
||||
#### Modified Files and Checklist
|
||||
|
||||
- [ ] `apps/edge/internal/openai/stream_gate_filters.go` — add the private compatibility bound and apply it to both repeat hold requirements only.
|
||||
- [ ] `apps/edge/internal/openai/stream_gate_filters_test.go` — add `TestOpenAIRepeatHoldBufferContract` covering participant, composed-plan, evidence-window, and schema non-regression assertions.
|
||||
- [ ] `agent-task/provider_stream_buffer_compat/CODE_REVIEW-cloud-G05.md` — record implementation decisions and actual item verification output.
|
||||
|
||||
#### Test Strategy
|
||||
|
||||
Write a regression test because this is a bug fix. `TestOpenAIRepeatHoldBufferContract` must use production registration/resolution rather than only direct constructor calls so omission of the automatically registered action sibling fails the test. It must assert exact rune limits and preserve the schema boundary explicitly.
|
||||
|
||||
#### Verification
|
||||
|
||||
```bash
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_filters.go apps/edge/internal/openai/stream_gate_filters_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^Test(OpenAIRepeatHoldBufferContract|OpenAIOutputFilterRegistrations)$'
|
||||
```
|
||||
|
||||
Expected: no formatting diff; the fresh focused tests pass and prove the repeat pair composes to 1 MiB while its evidence window and schema bound remain unchanged.
|
||||
|
||||
### [STREAM_BUFFER-2] Large semantic-delta exact-wire tunnel regression
|
||||
|
||||
#### Problem
|
||||
|
||||
`apps/edge/internal/openai/stream_gate_tunnel_codec.go:239-286,323-408,444-556` parses a complete SSE frame into one content or reasoning event and retains the original frame for release. `packages/go/streamgate/evidence_tail.go:1569-1582` checks the event's rune count against the composed hold bound before evaluation. Existing exact-wire fixtures at `apps/edge/internal/openai/stream_gate_pipeline_test.go:375-443` use only small deltas, so they do not reproduce the valid greater-than-4,096 event that currently terminates as a 502.
|
||||
|
||||
Before:
|
||||
|
||||
```go
|
||||
// apps/edge/internal/openai/stream_gate_pipeline_test.go:375-443
|
||||
func TestOpenAITunnelCodecTerminalWire(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
endpoint string
|
||||
frames [][]byte
|
||||
}{
|
||||
{
|
||||
name: "chat finish then done",
|
||||
endpoint: openAIRebuildEndpointChat,
|
||||
frames: [][]byte{
|
||||
[]byte("data: {\"choices\":[{\"delta\":{\"content\":\"answer\"}}]}\n\n"),
|
||||
// ...
|
||||
},
|
||||
},
|
||||
// Responses and transport-end variants also use small payloads.
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### Solution
|
||||
|
||||
- Add `TestStreamGateConfiguredRepeatGuardLargeTunnelDelta` beside the production tunnel lifecycle tests.
|
||||
- Use a table with four cases: Chat content, Chat reasoning, Responses output text, and Responses reasoning. Each case must encode one SSE JSON frame containing 5,000 distinct Korean runes, followed by the endpoint terminal wire.
|
||||
- Configure only blocking `repeat_guard` with the normal 500-rune evidence window and no schema metadata. Build the production registry, real tunnel event source/Core runtime, and real tunnel release sink over buffered `ProviderTunnelFrame` fixtures.
|
||||
- For every case, require runtime success, one HTTP 200 start, one successful terminal, zero recovery dispatches, no `provider_tunnel_error`/`buffer_overflow`, and caller body bytes exactly equal to the concatenated provider frames.
|
||||
- Keep the provider frame whole. Do not add codec fragmentation or a replacement serializer.
|
||||
|
||||
After:
|
||||
|
||||
```go
|
||||
func TestStreamGateConfiguredRepeatGuardLargeTunnelDelta(t *testing.T) {
|
||||
payload := uniqueKoreanRunes(5000)
|
||||
tests := []struct {
|
||||
name string
|
||||
endpoint string
|
||||
frame []byte
|
||||
}{
|
||||
{name: "chat content", endpoint: openAIRebuildEndpointChat, frame: chatContentFrame(payload)},
|
||||
{name: "chat reasoning", endpoint: openAIRebuildEndpointChat, frame: chatReasoningFrame(payload)},
|
||||
{name: "responses output text", endpoint: openAIRebuildEndpointResponses, frame: responsesTextFrame(payload)},
|
||||
{name: "responses reasoning", endpoint: openAIRebuildEndpointResponses, frame: responsesReasoningFrame(payload)},
|
||||
}
|
||||
// Each case runs production registry -> endpoint codec -> Core -> tunnel sink
|
||||
// and compares the caller body with the original semantic + terminal frames.
|
||||
}
|
||||
```
|
||||
|
||||
Use local fixture construction with `encoding/json` or existing imports; do not add a package dependency. Helper names in the outline are illustrative and may be replaced by a compact local frame builder within the same test.
|
||||
|
||||
#### Modified Files and Checklist
|
||||
|
||||
- [ ] `apps/edge/internal/openai/stream_gate_pipeline_test.go` — add the four-case production runtime regression and exact-wire assertions.
|
||||
- [ ] `agent-task/provider_stream_buffer_compat/CODE_REVIEW-cloud-G05.md` — record the actual regression output and any justified deviation.
|
||||
|
||||
#### Test Strategy
|
||||
|
||||
Write the integration regression in the existing pipeline test file. The 5,000-rune distinct payload is above the old default and below the new bound, avoids accidentally triggering repeat detection, exercises UTF-8 rune rather than byte accounting, and covers both endpoint and semantic-kind axes.
|
||||
|
||||
#### Verification
|
||||
|
||||
```bash
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_pipeline_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^TestStreamGateConfiguredRepeatGuardLargeTunnelDelta$'
|
||||
```
|
||||
|
||||
Expected: no formatting diff; all four fresh subtests return the original wire once with HTTP 200 and a successful terminal.
|
||||
|
||||
## Modified Files Summary
|
||||
|
||||
| File | Items |
|
||||
|------|-------|
|
||||
| `apps/edge/internal/openai/stream_gate_filters.go` | STREAM_BUFFER-1 |
|
||||
| `apps/edge/internal/openai/stream_gate_filters_test.go` | STREAM_BUFFER-1 |
|
||||
| `apps/edge/internal/openai/stream_gate_pipeline_test.go` | STREAM_BUFFER-2 |
|
||||
| `agent-task/provider_stream_buffer_compat/CODE_REVIEW-cloud-G05.md` | STREAM_BUFFER-1, STREAM_BUFFER-2 implementation evidence |
|
||||
|
||||
## Dependencies and Execution Order
|
||||
|
||||
1. Implement `STREAM_BUFFER-1` so the production registry resolves the corrected composed hold bound.
|
||||
2. Implement `STREAM_BUFFER-2` against that production registry and retain exact provider wire.
|
||||
3. Run the focused commands, then the complete final verification, and fill the active review evidence.
|
||||
|
||||
## Final Verification
|
||||
|
||||
```bash
|
||||
command -v go && readlink -f "$(command -v go)" && go version && go env GOROOT && go env GOMOD
|
||||
make proto
|
||||
git diff --exit-code -- proto/gen/iop
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_filters.go apps/edge/internal/openai/stream_gate_filters_test.go apps/edge/internal/openai/stream_gate_pipeline_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^Test(OpenAIRepeatHoldBufferContract|StreamGateConfiguredRepeatGuardLargeTunnelDelta)$'
|
||||
go test -race -count=1 ./packages/go/streamgate ./apps/edge/internal/openai ./packages/go/config
|
||||
make test-openai-lemonade
|
||||
IOP_E2E_BIND_TIMEOUT=60 IOP_E2E_PROFILE=mock ./scripts/e2e-smoke.sh
|
||||
git diff --check
|
||||
```
|
||||
|
||||
Expected: the host Go identity remains unchanged; proto generation produces no diff; formatting is clean; fresh focused and race suites pass; the repository-native OpenAI provider tunnel and mock smokes pass; and the final diff has no whitespace errors. No live provider, external runner, or cached Go result is accepted as a substitute for the deterministic large-delta regression.
|
||||
|
||||
After completing all code changes, fill implementation-owned sections in `CODE_REVIEW-*-G??.md`.
|
||||
|
|
@ -0,0 +1,276 @@
|
|||
<!-- task=provider_stream_buffer_compat plan=1 tag=STREAM_BUFFER -->
|
||||
|
||||
# Provider stream single-event buffer compatibility hardening
|
||||
|
||||
## For the Implementing Agent
|
||||
|
||||
Filling the implementation-owned sections in `CODE_REVIEW-*-G??.md` is mandatory. Run every verification command, paste actual notes and stdout/stderr into the active review file, keep the active PLAN/CODE_REVIEW files in place, and report ready for review; only the code-review skill may finalize or archive the task. If blocked, record only the exact blocker, attempted commands/output, and resume conditions in implementation-owned evidence fields. Do not ask the user, call user-input tools, create control-plane stop files, classify the next state, archive logs, or write `complete.log`.
|
||||
|
||||
## Background
|
||||
|
||||
The current branch's configured repeat policy inherits Core's 4,096-rune default hard buffer for both its rolling text guard and terminal action sibling. A provider tunnel SSE frame can decode into one valid content or reasoning event larger than that default, causing Core to emit `buffer_overflow` before repeat evaluation and Edge to return a 502 `provider_tunnel_error`. This patch changes only the repeat policy's infrastructure hold envelope to Core's existing 1,048,576-rune ceiling; it does not change provider output-filter decisions or provider wire bytes.
|
||||
|
||||
## Archive Evidence Snapshot
|
||||
|
||||
- The user requested a second plan review before implementation. The prior unimplemented draft pair is preserved as `agent-task/provider_stream_buffer_compat/plan_cloud_G05_0.log` and `agent-task/provider_stream_buffer_compat/code_review_cloud_G05_0.log`.
|
||||
- The prior review stub has no verdict, implementation evidence, Required/Suggested/Nit findings, or verification result. This replan retains the diagnosed two-requirement bound fix and production-path regression, corrects the unit from “MiB” to runes, removes a stale prior-task reference, and removes non-diagnostic proto/general smoke commands.
|
||||
- No roadmap state or completion claim carries over.
|
||||
|
||||
## Analysis
|
||||
|
||||
### Files Read
|
||||
|
||||
- Workflow and rules: `AGENTS.md`, `agent-ops/rules/project/rules.md`, `agent-ops/rules/private/rules.md`, `agent-ops/rules/common/rules-roadmap.md`, `agent-ops/rules/common/rules-agent-spec.md`, `agent-ops/rules/project/domain/edge/rules.md`, `agent-ops/rules/project/domain/platform-common/rules.md`, `agent-ops/rules/project/domain/testing/rules.md`, `agent-ops/skills/common/router.md`, `agent-ops/skills/common/plan/SKILL.md`, `agent-ops/skills/common/finalize-task-routing/SKILL.md`, and `agent-ops/skills/common/plan/templates/review-stub-template.md`.
|
||||
- Verification profiles: `agent-test/local/rules.md`, `agent-test/local/edge-smoke.md`, `agent-test/local/platform-common-smoke.md`, and `agent-test/local/testing-smoke.md`.
|
||||
- Contracts and current specs: `agent-contract/index.md`, `agent-contract/outer/openai-compatible-api.md`, `agent-contract/inner/edge-node-runtime-wire.md`, `agent-spec/index.md`, `agent-spec/runtime/stream-evidence-gate.md`, and `agent-spec/input/openai-compatible-surface.md`.
|
||||
- Edge source: `apps/edge/internal/openai/stream_gate_filters.go`, `apps/edge/internal/openai/stream_gate_policy.go`, `apps/edge/internal/openai/stream_gate_tunnel_codec.go`, `apps/edge/internal/openai/stream_gate_release_sink.go`, and `apps/edge/internal/openai/stream_gate_runtime.go`.
|
||||
- Core source: `packages/go/streamgate/evidence_tail.go`, `packages/go/streamgate/runtime.go`, and `packages/go/streamgate/filter_registry.go`.
|
||||
- Tests: `apps/edge/internal/openai/stream_gate_filters_test.go`, `apps/edge/internal/openai/stream_gate_policy_test.go`, `apps/edge/internal/openai/stream_gate_pipeline_test.go`, and `packages/go/streamgate/stream_release_test.go`.
|
||||
- Prior task-local draft: `agent-task/provider_stream_buffer_compat/plan_cloud_G05_0.log` and `agent-task/provider_stream_buffer_compat/code_review_cloud_G05_0.log`.
|
||||
|
||||
### SDD Criteria
|
||||
|
||||
Not applicable. This is a non-roadmap compatibility patch and must not check or modify any Milestone Task or SDD state.
|
||||
|
||||
### Verification Context
|
||||
|
||||
- No `verification_context` handoff was supplied. Repository-native evidence is the current OpenAI contract/spec, Core hold implementation, production Edge tunnel codec/runtime/release sink, existing exact-wire tests, and local test rules.
|
||||
- Local preflight on 2026-07-30: branch `feature/openai-compatible-output-validation-filters`, HEAD `213eee4e28dfa69ff1e412faf23978ee9a9a3b9f`; Go resolves from `/config/.local/bin/go` to `/config/opt/go/bin/go`, version `go1.26.2 linux/arm64`, `GOROOT=/config/opt/go`, `GOMOD=/config/workspace/iop/go.mod`.
|
||||
- The worktree has user-owned deletions under `agent-task/m-openai-compatible-output-validation-filters/02+01_repeat_guard/` and this task's plan artifacts. Preserve those deletions; no planned source file has a diff. No dispatcher process is running.
|
||||
- Existing focused harness proof passed uncached: `go test -count=1 ./apps/edge/internal/openai -run 'Test(RepeatGuardIdleDoesNotRelease|OpenAIOutputFilterRegistrations|OpenAITunnelCodecTerminalWire|StreamGateConfiguredRepeatActionSplitLifecycle)$'`.
|
||||
- No external runner, provider, shared runtime, dev deployment, credential, network endpoint, protobuf generation, or config migration is required. The decisive oracle is an in-process production registry + tunnel event source + Core runtime + tunnel release sink driven by deterministic provider frames.
|
||||
- General mock/provider smoke commands do not enable `stream_evidence_gate` with blocking `repeat_guard` and cannot distinguish the old 4,096-rune failure from the fix. They are not acceptance evidence for this patch; the production-path regression below is.
|
||||
- Fresh `-count=1` and `-race -count=1` results are required; cached Go test output is not acceptable. Confidence is high because the regression crosses the exact production codec/Core/sink boundary.
|
||||
|
||||
### Test Coverage Gaps
|
||||
|
||||
- Core tests cover explicit `max_buffer_runes`, Unicode rune counting, overflow, discard, and terminal behavior, but no Edge test locks the bound produced by the repeat text guard plus its automatically registered action sibling.
|
||||
- Existing tunnel tests cover Chat/Responses event shapes, terminal framing, split tool identity, and byte-identical release only with small semantic deltas.
|
||||
- No current test drives one content or reasoning event larger than 4,096 runes through the configured repeat policy and complete tunnel runtime. `STREAM_BUFFER-1` closes the policy-composition gap; `STREAM_BUFFER-2` closes the observed transport regression gap.
|
||||
|
||||
### Symbol References
|
||||
|
||||
None. No symbol is renamed or removed, and no public API or dependency is introduced.
|
||||
|
||||
### Split Judgment
|
||||
|
||||
Keep one compact plan. The indivisible invariant is that a repeat-enabled provider tunnel accepts one valid coalesced semantic event within Core's existing ceiling and releases the original SSE bytes exactly once; the hold configuration and runtime regression cannot independently prove that invariant.
|
||||
|
||||
### Scope Rationale
|
||||
|
||||
- Change only the repeat policy's rolling text/reasoning hold and terminal action sibling hold. Both need the same bound because Core composes blocking requirements on one channel using the minimum positive `max_buffer_runes`.
|
||||
- Preserve `hold_evidence_runes` as the repeat evaluation window. The new hard bound is buffer compatibility, not a larger repeat-detection window.
|
||||
- Raising the action sibling's bound also raises the channel-level allowance for pending tool-call fragments. This is an unavoidable consequence of the current channel-level Core composition, remains bounded at 1,048,576 runes, and does not change action evaluation or release rules.
|
||||
- Preserve the schema gate's default 4,096-rune terminal bound. A request with `metadata.scheme` is a distinct full-output validation boundary and is not claimed fixed by this communication patch.
|
||||
- Do not change `packages/go/streamgate`, codec framing/parsing, release queue semantics, 502 serialization, config schema, contracts, specs, roadmap, deployment, or provider output-filter evaluation behavior.
|
||||
- Do not split or reserialize provider frames. Core already buffers and releases a normalized event as one entry; codec fragmentation would require group-aware release semantics and is outside this fix.
|
||||
- Events above 1,048,576 runes remain intentionally fail-closed.
|
||||
|
||||
### Final Routing
|
||||
|
||||
- `status=routed`; `evaluation_mode=isolated-reassessment`; `finalizer=finalize-task-policy.sh`; `finalizer_mode=pair`.
|
||||
- Build closures: scope/context/verification/evidence/ownership/decision closed. Grade scores `1/1/1/1/1`, base basis `local-fit`, final route basis `risk-boundary`, lane `cloud`, grade `G05`, canonical file `PLAN-cloud-G05.md`.
|
||||
- Review closures: scope/context/verification/evidence/ownership/decision closed. Grade scores `1/1/1/1/1`, route `official-review`, lane `cloud`, grade `G05`, adapter `codex`, model `gpt-5.6-sol`, reasoning effort `xhigh`, canonical file `CODE_REVIEW-cloud-G05.md`.
|
||||
- `large_indivisible_context=false`.
|
||||
- Positive loop risks: `temporal_state`, `boundary_contract`, `structured_interpretation`, and `variant_product`; count `4`.
|
||||
- Recovery signals: `review_rework_count=0`, `evidence_integrity_failure=false`; risk boundary matched, recovery boundary did not.
|
||||
- Capability gap: none.
|
||||
|
||||
## Implementation Checklist
|
||||
|
||||
- [ ] [STREAM_BUFFER-1] Align both repeat-policy hold requirements to Core's existing 1,048,576-rune ceiling and lock their composed envelope with a unit regression.
|
||||
- [ ] [STREAM_BUFFER-2] Add Chat/Responses content/reasoning tunnel regressions for one semantic event above 4,096 runes with byte-identical successful release.
|
||||
- [ ] Fill implementation-owned sections in CODE_REVIEW-*-G??.md with actual implementation notes and verification output.
|
||||
|
||||
### [STREAM_BUFFER-1] Repeat-policy hold envelope
|
||||
|
||||
#### Problem
|
||||
|
||||
`apps/edge/internal/openai/stream_gate_filters.go:131-173` builds the repeat rolling guard and terminal action sibling with default constructors. Both inherit the 4,096-rune default from `packages/go/streamgate/evidence_tail.go:74-81,157-197`. `packages/go/streamgate/evidence_tail.go:448-483` composes same-channel blocking requirements using the minimum positive bound, so changing only one repeat participant leaves 4,096 effective.
|
||||
|
||||
Before:
|
||||
|
||||
```go
|
||||
// apps/edge/internal/openai/stream_gate_filters.go:131-173
|
||||
case openAIOutputFilterRepeatGuard:
|
||||
req, err := streamgate.NewFilterHoldRequirementRolling(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{
|
||||
streamgate.EventKindTextDelta,
|
||||
streamgate.EventKindReasoningDelta,
|
||||
},
|
||||
f.holdRunes,
|
||||
)
|
||||
// ...
|
||||
case openAIOutputFilterRepeatActionGuard:
|
||||
req, err := streamgate.NewFilterHoldRequirementTerminalGate(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{
|
||||
streamgate.EventKindToolCallFragment,
|
||||
streamgate.EventKindTerminal,
|
||||
},
|
||||
streamgate.EventKindTerminal,
|
||||
)
|
||||
```
|
||||
|
||||
#### Solution
|
||||
|
||||
- Add private Edge constant `openAIRepeatHoldMaxBufferRunes = 1 << 20`, documented as a coalesced semantic-event allowance measured in runes, not bytes and not the rolling evidence window.
|
||||
- Use `NewFilterHoldRequirementRollingWithMaxBuffer` for `openAIOutputFilterRepeatGuard` and `NewFilterHoldRequirementTerminalGateWithMaxBuffer` for `openAIOutputFilterRepeatActionGuard`, including their defensive fallback calls.
|
||||
- Leave `openAIOutputFilterSchemaGate` on the default constructor.
|
||||
- Add `TestOpenAIRepeatHoldBufferContract`: resolve the production repeat registration without schema, compile `streamgate.EvidencePlan`, and assert the composed channel maximum and both repeat participant maxima are 1,048,576 runes, the rolling evidence value remains configured, and a standalone schema requirement remains 4,096.
|
||||
|
||||
After:
|
||||
|
||||
```go
|
||||
// A provider may coalesce one valid content/reasoning delta above Core's
|
||||
// default hold size. This bound is measured in runes and does not widen evidence.
|
||||
const openAIRepeatHoldMaxBufferRunes = 1 << 20
|
||||
|
||||
case openAIOutputFilterRepeatGuard:
|
||||
req, err := streamgate.NewFilterHoldRequirementRollingWithMaxBuffer(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{
|
||||
streamgate.EventKindTextDelta,
|
||||
streamgate.EventKindReasoningDelta,
|
||||
},
|
||||
f.holdRunes,
|
||||
openAIRepeatHoldMaxBufferRunes,
|
||||
)
|
||||
// fallback uses the same explicit bound
|
||||
return req
|
||||
case openAIOutputFilterRepeatActionGuard:
|
||||
req, err := streamgate.NewFilterHoldRequirementTerminalGateWithMaxBuffer(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{
|
||||
streamgate.EventKindToolCallFragment,
|
||||
streamgate.EventKindTerminal,
|
||||
},
|
||||
streamgate.EventKindTerminal,
|
||||
openAIRepeatHoldMaxBufferRunes,
|
||||
)
|
||||
// fallback uses the same explicit bound
|
||||
return req
|
||||
case openAIOutputFilterSchemaGate:
|
||||
// Keep the existing default constructor and bound.
|
||||
```
|
||||
|
||||
#### Modified Files and Checklist
|
||||
|
||||
- [ ] `apps/edge/internal/openai/stream_gate_filters.go` — add the private rune bound and apply it to both repeat hold requirements only.
|
||||
- [ ] `apps/edge/internal/openai/stream_gate_filters_test.go` — add `TestOpenAIRepeatHoldBufferContract` for participant, composed-plan, evidence-window, and schema non-regression assertions.
|
||||
- [ ] `agent-task/provider_stream_buffer_compat/CODE_REVIEW-cloud-G05.md` — record actual implementation decisions and verification output.
|
||||
|
||||
#### Test Strategy
|
||||
|
||||
Write a regression because this is a bug fix. The test must use production registration/resolution rather than only direct constructors, so omission of the automatically registered action sibling fails.
|
||||
|
||||
#### Verification
|
||||
|
||||
```bash
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_filters.go apps/edge/internal/openai/stream_gate_filters_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^Test(OpenAIRepeatHoldBufferContract|OpenAIOutputFilterRegistrations)$'
|
||||
```
|
||||
|
||||
Expected: no formatting diff; fresh tests prove the repeat pair composes to 1,048,576 runes while the evidence window and schema bound remain unchanged.
|
||||
|
||||
### [STREAM_BUFFER-2] Large semantic-event exact-wire tunnel regression
|
||||
|
||||
#### Problem
|
||||
|
||||
`apps/edge/internal/openai/stream_gate_tunnel_codec.go:239-286` converts a complete SSE frame into one or more semantic events and attaches the original wire frame to the first releasable event. `packages/go/streamgate/evidence_tail.go:1564-1582` rejects the whole event before evaluation when its rune count exceeds the composed hold bound. Existing production lifecycle coverage at `apps/edge/internal/openai/stream_gate_pipeline_test.go:612-720` uses small payloads and does not reproduce the observed greater-than-4,096 event.
|
||||
|
||||
Before:
|
||||
|
||||
```go
|
||||
// apps/edge/internal/openai/stream_gate_pipeline_test.go:612-634
|
||||
func TestStreamGateConfiguredRepeatActionSplitLifecycle(t *testing.T) {
|
||||
// Production registry/Core/tunnel coverage exists, but its frames are small.
|
||||
gateCfg := config.StreamEvidenceGateConf{
|
||||
Enabled: true,
|
||||
Filters: []config.StreamGateFilterPolicyConf{{
|
||||
Filter: config.StreamGateFilterRepeatGuard,
|
||||
Enforcement: config.StreamGateFilterEnforcementBlocking,
|
||||
HoldEvidenceRunes: 500,
|
||||
}},
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### Solution
|
||||
|
||||
- Add `TestStreamGateConfiguredRepeatGuardLargeTunnelEvent` beside the production tunnel lifecycle tests.
|
||||
- Use four cases: Chat content (`choices[].delta.content`), Chat reasoning (`choices[].delta.reasoning_content`), Responses output text (`response.output_text.delta`), and Responses reasoning (`response.reasoning_text.delta`).
|
||||
- Encode one SSE JSON frame with 5,000 distinct Korean runes per case, followed by endpoint-native terminal wire. The payload is above the old default, below the new bound, and cannot trigger repeat matching.
|
||||
- Configure only blocking `repeat_guard` with the normal 500-rune evidence window and no schema metadata. Drive the production registry, tunnel event source, Core runtime, and tunnel release sink with buffered `ProviderTunnelFrame` fixtures.
|
||||
- Require `runtime.Run` success, one HTTP 200 start, one successful terminal, zero recovery dispatches, and caller body bytes exactly equal to the concatenated provider frames.
|
||||
- Keep each provider frame whole; do not add codec fragmentation or a replacement serializer.
|
||||
|
||||
After:
|
||||
|
||||
```go
|
||||
func TestStreamGateConfiguredRepeatGuardLargeTunnelEvent(t *testing.T) {
|
||||
payload := uniqueKoreanRunes(5000)
|
||||
tests := []struct {
|
||||
name string
|
||||
endpoint string
|
||||
frame []byte
|
||||
}{
|
||||
{name: "chat content", endpoint: openAIRebuildEndpointChat, frame: chatContentFrame(payload)},
|
||||
{name: "chat reasoning", endpoint: openAIRebuildEndpointChat, frame: chatReasoningFrame(payload)},
|
||||
{name: "responses output text", endpoint: openAIRebuildEndpointResponses, frame: responsesTextFrame(payload)},
|
||||
{name: "responses reasoning", endpoint: openAIRebuildEndpointResponses, frame: responsesReasoningFrame(payload)},
|
||||
}
|
||||
// Each case runs production registry -> tunnel source -> Core -> tunnel sink
|
||||
// and compares caller bytes with the original semantic and terminal frames.
|
||||
}
|
||||
```
|
||||
|
||||
Helper names are illustrative. Build the JSON/SSE fixtures locally with `encoding/json` or existing imports; add no dependency.
|
||||
|
||||
#### Modified Files and Checklist
|
||||
|
||||
- [ ] `apps/edge/internal/openai/stream_gate_pipeline_test.go` — add the four-case production runtime regression and exact-wire assertions.
|
||||
- [ ] `agent-task/provider_stream_buffer_compat/CODE_REVIEW-cloud-G05.md` — record actual regression output and any justified deviation.
|
||||
|
||||
#### Test Strategy
|
||||
|
||||
Write the integration regression in the existing pipeline test file. The distinct 5,000-rune payload exercises UTF-8 rune accounting and both endpoint and semantic-kind axes without changing filter semantics.
|
||||
|
||||
#### Verification
|
||||
|
||||
```bash
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_pipeline_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^TestStreamGateConfiguredRepeatGuardLargeTunnelEvent$'
|
||||
```
|
||||
|
||||
Expected: no formatting diff; all four fresh subtests return the original wire once with HTTP 200, one successful terminal, and no recovery dispatch.
|
||||
|
||||
## Modified Files Summary
|
||||
|
||||
| File | Items |
|
||||
|------|-------|
|
||||
| `apps/edge/internal/openai/stream_gate_filters.go` | STREAM_BUFFER-1 |
|
||||
| `apps/edge/internal/openai/stream_gate_filters_test.go` | STREAM_BUFFER-1 |
|
||||
| `apps/edge/internal/openai/stream_gate_pipeline_test.go` | STREAM_BUFFER-2 |
|
||||
| `agent-task/provider_stream_buffer_compat/CODE_REVIEW-cloud-G05.md` | STREAM_BUFFER-1, STREAM_BUFFER-2 implementation evidence |
|
||||
|
||||
## Dependencies and Execution Order
|
||||
|
||||
1. Implement `STREAM_BUFFER-1` so the production registry resolves the corrected composed bound.
|
||||
2. Implement `STREAM_BUFFER-2` against that registry and retain exact provider wire.
|
||||
3. Run focused and final verification, then fill the active review evidence.
|
||||
|
||||
## Final Verification
|
||||
|
||||
```bash
|
||||
command -v go && readlink -f "$(command -v go)" && go version && go env GOROOT && go env GOMOD
|
||||
gofmt -d apps/edge/internal/openai/stream_gate_filters.go apps/edge/internal/openai/stream_gate_filters_test.go apps/edge/internal/openai/stream_gate_pipeline_test.go
|
||||
go test -count=1 ./apps/edge/internal/openai -run '^Test(OpenAIRepeatHoldBufferContract|StreamGateConfiguredRepeatGuardLargeTunnelEvent)$'
|
||||
go test -race -count=1 ./packages/go/streamgate ./apps/edge/internal/openai ./packages/go/config
|
||||
git diff --check
|
||||
```
|
||||
|
||||
Expected: host Go identity is unchanged; formatting is clean; fresh focused and race suites pass; all four large-event tunnel variants preserve the exact provider wire with one successful terminal; and the diff has no whitespace errors. Proto generation and general mock/provider smoke are excluded because no proto/config/route is changed and those commands do not activate this repeat-policy failure condition.
|
||||
|
||||
After completing all code changes, fill implementation-owned sections in `CODE_REVIEW-*-G??.md`.
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
# Milestone Work Log
|
||||
|
||||
> Dispatcher-owned execution timeline. Workers and reviewers do not edit this file.
|
||||
|
||||
| seq | time | event | task | role | attempt | model | result | locator |
|
||||
|---:|---|---|---|---|---:|---|---|---|
|
||||
| 1 | 26-07-30 06:34:32 | START | provider_stream_buffer_compat | worker | 0 | agy/Gemini 3.6 Flash (High) | running | /config/workspace/iop/.git/agent-task-dispatcher/runs/20260729T213432Z__provider_stream_buffer_compat__p1__worker__a00/locator.json |
|
||||
| 2 | 26-07-30 06:37:03 | FINISH | provider_stream_buffer_compat | worker | 0 | agy/Gemini 3.6 Flash (High) | succeeded:0 | /config/workspace/iop/.git/agent-task-dispatcher/runs/20260729T213432Z__provider_stream_buffer_compat__p1__worker__a00/locator.json |
|
||||
| 3 | 26-07-30 06:37:03 | START | provider_stream_buffer_compat | review | 0 | codex/gpt-5.6-sol xhigh | running | /config/workspace/iop/.git/agent-task-dispatcher/runs/20260729T213703Z__provider_stream_buffer_compat__p1__review__a00/locator.json |
|
||||
| 4 | 26-07-30 06:43:07 | FINISH | provider_stream_buffer_compat | review | 0 | codex/gpt-5.6-sol xhigh | succeeded:0 | /config/workspace/iop/.git/agent-task-dispatcher/runs/20260729T213703Z__provider_stream_buffer_compat__p1__review__a00/locator.json |
|
||||
|
|
@ -122,6 +122,10 @@ func newOpenAIOutputFilter(kind openAIOutputFilterKind, holdRunes, priority int,
|
|||
return filter, nil
|
||||
}
|
||||
|
||||
// A provider may coalesce one valid content/reasoning delta above Core's
|
||||
// default hold size. This bound is measured in runes and does not widen evidence.
|
||||
const openAIRepeatHoldMaxBufferRunes = 1 << 20
|
||||
|
||||
// Applies is execution-path-neutral because endpoint codecs expose the same
|
||||
// semantic event kinds for normalized and provider-tunnel attempts.
|
||||
func (f *openAIOutputFilter) Applies(streamgate.FilterContext) bool {
|
||||
|
|
@ -132,32 +136,40 @@ func (f *openAIOutputFilter) Applies(streamgate.FilterContext) bool {
|
|||
func (f *openAIOutputFilter) HoldRequirement(streamgate.FilterContext) streamgate.FilterHoldRequirement {
|
||||
switch f.kind {
|
||||
case openAIOutputFilterRepeatGuard:
|
||||
req, err := streamgate.NewFilterHoldRequirementRolling(
|
||||
req, err := streamgate.NewFilterHoldRequirementRollingWithMaxBuffer(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{
|
||||
streamgate.EventKindTextDelta,
|
||||
streamgate.EventKindReasoningDelta,
|
||||
},
|
||||
f.holdRunes,
|
||||
openAIRepeatHoldMaxBufferRunes,
|
||||
)
|
||||
if err != nil {
|
||||
req, _ = streamgate.NewFilterHoldRequirementRolling(f.channel, []streamgate.EventKind{streamgate.EventKindTextDelta}, f.holdRunes)
|
||||
req, _ = streamgate.NewFilterHoldRequirementRollingWithMaxBuffer(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{streamgate.EventKindTextDelta},
|
||||
f.holdRunes,
|
||||
openAIRepeatHoldMaxBufferRunes,
|
||||
)
|
||||
}
|
||||
return req
|
||||
case openAIOutputFilterRepeatActionGuard:
|
||||
req, err := streamgate.NewFilterHoldRequirementTerminalGate(
|
||||
req, err := streamgate.NewFilterHoldRequirementTerminalGateWithMaxBuffer(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{
|
||||
streamgate.EventKindToolCallFragment,
|
||||
streamgate.EventKindTerminal,
|
||||
},
|
||||
streamgate.EventKindTerminal,
|
||||
openAIRepeatHoldMaxBufferRunes,
|
||||
)
|
||||
if err != nil {
|
||||
req, _ = streamgate.NewFilterHoldRequirementTerminalGate(
|
||||
req, _ = streamgate.NewFilterHoldRequirementTerminalGateWithMaxBuffer(
|
||||
f.channel,
|
||||
[]streamgate.EventKind{streamgate.EventKindToolCallFragment},
|
||||
streamgate.EventKindTerminal,
|
||||
openAIRepeatHoldMaxBufferRunes,
|
||||
)
|
||||
}
|
||||
return req
|
||||
|
|
|
|||
|
|
@ -1037,3 +1037,106 @@ func TestOpenAIRepeatAndSchemaFiltersPassCleanEpoch(t *testing.T) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIRepeatHoldBufferContract(t *testing.T) {
|
||||
gateCfg := outputFilterGateCfg(config.StreamGateFilterEnforcementBlocking)
|
||||
fctxNoScheme := openAIOutputFilterContext{endpoint: openAIRebuildEndpointChat, requestRef: "openai.snap.1"}
|
||||
regsNoScheme, policiesNoScheme, err := openAIOutputFilterRegistrations(gateCfg, fctxNoScheme)
|
||||
if err != nil {
|
||||
t.Fatalf("openAIOutputFilterRegistrations(no scheme): %v", err)
|
||||
}
|
||||
|
||||
snap, err := streamgate.NewFilterRegistrySnapshot(streamGateConfigGeneration, regsNoScheme, policiesNoScheme)
|
||||
if err != nil {
|
||||
t.Fatalf("NewFilterRegistrySnapshot: %v", err)
|
||||
}
|
||||
reqCtx, err := streamgate.NewRequestFilterContext(
|
||||
streamGateConfigGeneration, "attempt.1", streamGateEnvironment,
|
||||
openAIRebuildEndpointChat, openAIRebuildFamily, "",
|
||||
streamgate.CommitStateTransportUncommitted, false, false, "",
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("NewRequestFilterContext: %v", err)
|
||||
}
|
||||
reqSnap, err := snap.BeginRequest(reqCtx)
|
||||
if err != nil {
|
||||
t.Fatalf("BeginRequest: %v", err)
|
||||
}
|
||||
target, err := streamgate.NewAttemptTarget("client-model", "ornith:35b", "prov-a", "normalized",
|
||||
[]string{"output.repeat_guard", "output.provider_error"})
|
||||
if err != nil {
|
||||
t.Fatalf("NewAttemptTarget: %v", err)
|
||||
}
|
||||
resolved, err := reqSnap.ResolveAttempt(target)
|
||||
if err != nil {
|
||||
t.Fatalf("ResolveAttempt: %v", err)
|
||||
}
|
||||
|
||||
plan, err := streamgate.NewEvidencePlanFromResolvedFilters(resolved)
|
||||
if err != nil {
|
||||
t.Fatalf("NewEvidencePlanFromResolvedFilters: %v", err)
|
||||
}
|
||||
|
||||
const wantBound = openAIRepeatHoldMaxBufferRunes // 1048576
|
||||
if got := plan.MaxBufferRunes(streamGateChannelDefault); got != wantBound {
|
||||
t.Errorf("composed channel maxBufferRunes = %d, want %d", got, wantBound)
|
||||
}
|
||||
|
||||
bindings := plan.BindingsForChannel(streamGateChannelDefault)
|
||||
foundRepeat := false
|
||||
foundAction := false
|
||||
for _, b := range bindings {
|
||||
req := b.Requirement()
|
||||
switch b.FilterID() {
|
||||
case openAIRepeatGuardFilterID:
|
||||
foundRepeat = true
|
||||
if got := req.MaxBufferRunes(); got != wantBound {
|
||||
t.Errorf("repeat_guard MaxBufferRunes = %d, want %d", got, wantBound)
|
||||
}
|
||||
if got := req.EvidenceRunes(); got != 500 {
|
||||
t.Errorf("repeat_guard EvidenceRunes = %d, want 500", got)
|
||||
}
|
||||
case openAIRepeatActionGuardFilterID:
|
||||
foundAction = true
|
||||
if got := req.MaxBufferRunes(); got != wantBound {
|
||||
t.Errorf("repeat_action_guard MaxBufferRunes = %d, want %d", got, wantBound)
|
||||
}
|
||||
}
|
||||
}
|
||||
if !foundRepeat {
|
||||
t.Error("missing repeat_guard binding in compiled plan")
|
||||
}
|
||||
if !foundAction {
|
||||
t.Error("missing repeat_action_guard binding in compiled plan")
|
||||
}
|
||||
|
||||
// Verify standalone schema requirement retains default bound (4096).
|
||||
regsScheme, policiesScheme, err := openAIOutputFilterRegistrations(gateCfg, schemaOutputFilterContext("openai.snap.1"))
|
||||
if err != nil {
|
||||
t.Fatalf("openAIOutputFilterRegistrations(scheme): %v", err)
|
||||
}
|
||||
snapScheme, err := streamgate.NewFilterRegistrySnapshot(streamGateConfigGeneration, regsScheme, policiesScheme)
|
||||
if err != nil {
|
||||
t.Fatalf("NewFilterRegistrySnapshot(scheme): %v", err)
|
||||
}
|
||||
reqSnapScheme, err := snapScheme.BeginRequest(reqCtx)
|
||||
if err != nil {
|
||||
t.Fatalf("BeginRequest(scheme): %v", err)
|
||||
}
|
||||
targetScheme, err := streamgate.NewAttemptTarget("client-model", "ornith:35b", "prov-a", "normalized",
|
||||
[]string{"output.repeat_guard", "output.schema_gate", "output.provider_error"})
|
||||
if err != nil {
|
||||
t.Fatalf("NewAttemptTarget(scheme): %v", err)
|
||||
}
|
||||
resolvedScheme, err := reqSnapScheme.ResolveAttempt(targetScheme)
|
||||
if err != nil {
|
||||
t.Fatalf("ResolveAttempt(scheme): %v", err)
|
||||
}
|
||||
for _, rf := range resolvedScheme {
|
||||
if rf.FilterID() == openAISchemaGateFilterID {
|
||||
if got := rf.HoldRequirement().MaxBufferRunes(); got != streamgate.DefaultMaxBufferRunes {
|
||||
t.Errorf("schema_gate MaxBufferRunes = %d, want default %d", got, streamgate.DefaultMaxBufferRunes)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import (
|
|||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
edgeservice "iop/apps/edge/internal/service"
|
||||
"iop/packages/go/config"
|
||||
|
|
@ -1038,3 +1039,157 @@ func releaseTunnelCodecEvents(t *testing.T, state *openAITunnelCodecState, event
|
|||
}
|
||||
return w.body.String()
|
||||
}
|
||||
|
||||
func TestStreamGateConfiguredRepeatGuardLargeTunnelEvent(t *testing.T) {
|
||||
payload := uniqueKoreanRunes(5000)
|
||||
if got := utf8.RuneCountInString(payload); got != 5000 {
|
||||
t.Fatalf("uniqueKoreanRunes count = %d, want 5000", got)
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
endpoint string
|
||||
semanticFrame []byte
|
||||
terminalFrame []byte
|
||||
}{
|
||||
{
|
||||
name: "chat content",
|
||||
endpoint: openAIRebuildEndpointChat,
|
||||
semanticFrame: []byte(fmt.Sprintf("data: {\"id\":\"chatcmpl-1\",\"object\":\"chat.completion.chunk\",\"created\":1700000000,\"model\":\"served-model\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"%s\"}}]}\n\n", payload)),
|
||||
terminalFrame: []byte("data: [DONE]\n\n"),
|
||||
},
|
||||
{
|
||||
name: "chat reasoning",
|
||||
endpoint: openAIRebuildEndpointChat,
|
||||
semanticFrame: []byte(fmt.Sprintf("data: {\"id\":\"chatcmpl-1\",\"object\":\"chat.completion.chunk\",\"created\":1700000000,\"model\":\"served-model\",\"choices\":[{\"index\":0,\"delta\":{\"reasoning_content\":\"%s\"}}]}\n\n", payload)),
|
||||
terminalFrame: []byte("data: [DONE]\n\n"),
|
||||
},
|
||||
{
|
||||
name: "responses output text",
|
||||
endpoint: openAIRebuildEndpointResponses,
|
||||
semanticFrame: []byte(fmt.Sprintf("data: {\"type\":\"response.output_text.delta\",\"delta\":\"%s\"}\n\n", payload)),
|
||||
terminalFrame: []byte("data: [DONE]\n\n"),
|
||||
},
|
||||
{
|
||||
name: "responses reasoning",
|
||||
endpoint: openAIRebuildEndpointResponses,
|
||||
semanticFrame: []byte(fmt.Sprintf("data: {\"type\":\"response.reasoning_text.delta\",\"delta\":\"%s\"}\n\n", payload)),
|
||||
terminalFrame: []byte("data: [DONE]\n\n"),
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
fault := 3
|
||||
gateCfg := config.StreamEvidenceGateConf{
|
||||
Enabled: true,
|
||||
MaxRequestFaultRecovery: &fault,
|
||||
Filters: []config.StreamGateFilterPolicyConf{{
|
||||
Filter: config.StreamGateFilterRepeatGuard,
|
||||
Enforcement: config.StreamGateFilterEnforcementBlocking,
|
||||
Priority: 10,
|
||||
HoldEvidenceRunes: 500,
|
||||
}},
|
||||
}
|
||||
service := &providerFakeRunService{}
|
||||
srv := NewServer(config.EdgeOpenAIConf{
|
||||
Adapter: "openai-compat", Target: "served-model", TimeoutSec: 15,
|
||||
StreamEvidenceGate: gateCfg,
|
||||
}, service, nil)
|
||||
observations := &recordingOpenAIObservationSink{}
|
||||
srv.SetObservationSink(observations)
|
||||
|
||||
var rawReq []byte
|
||||
if tt.endpoint == openAIRebuildEndpointChat {
|
||||
rawReq = []byte(`{"model":"client-model","messages":[{"role":"user","content":"hi"}]}`)
|
||||
} else {
|
||||
rawReq = []byte(`{"model":"client-model","input":"hi"}`)
|
||||
}
|
||||
route := routeDispatch{Adapter: "openai-compat", Target: "served-model", TimeoutSec: 15}
|
||||
requestCtx := newTestRequestContext(t, route, rawReq)
|
||||
req := openAITunnelStreamGateRequest{
|
||||
route: route, ingress: requestCtx.ingress, endpoint: tt.endpoint,
|
||||
method: http.MethodPost, path: "/v1/" + tt.endpoint, stream: true,
|
||||
modelGroupKey: "client-model",
|
||||
authorize: func(context.Context) (map[string]string, error) { return nil, nil },
|
||||
rewriteBody: func(body []byte, _ string) ([]byte, error) { return body, nil },
|
||||
}
|
||||
frames := make(chan *iop.ProviderTunnelFrame)
|
||||
handle := &fakeTunnelHandle{
|
||||
dispatch: edgeservice.RunDispatch{
|
||||
RunID: "large-" + tt.name, ModelGroupKey: "client-model",
|
||||
Adapter: "openai-compat", Target: "served-model",
|
||||
ProviderID: "provider-a", ExecutionPath: string(edgeservice.ProviderPoolPathTunnel),
|
||||
},
|
||||
frames: frames,
|
||||
}
|
||||
fctx, err := srv.openAITunnelOutputFilterContext(req)
|
||||
if err != nil {
|
||||
t.Fatalf("openAITunnelOutputFilterContext: %v", err)
|
||||
}
|
||||
registry, err := openAIStreamGateRegistrySnapshotFor(gateCfg, fctx)
|
||||
if err != nil {
|
||||
t.Fatalf("openAIStreamGateRegistrySnapshotFor: %v", err)
|
||||
}
|
||||
writer := newSynchronizedTunnelLifecycleWriter()
|
||||
sink := newOpenAITunnelReleaseSink(writer, writer)
|
||||
runtime, _, err := srv.buildOpenAITunnelStreamGateRuntime(req, handle, sink, registry)
|
||||
if err != nil {
|
||||
t.Fatalf("buildOpenAITunnelStreamGateRuntime: %v", err)
|
||||
}
|
||||
runDone := make(chan error, 1)
|
||||
go func() {
|
||||
runDone <- runtime.Run(t.Context())
|
||||
}()
|
||||
|
||||
frames <- &iop.ProviderTunnelFrame{
|
||||
Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START,
|
||||
StatusCode: http.StatusOK,
|
||||
Headers: map[string]string{"Content-Type": "text/event-stream"},
|
||||
}
|
||||
frames <- &iop.ProviderTunnelFrame{
|
||||
Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY,
|
||||
Body: tt.semanticFrame,
|
||||
}
|
||||
frames <- &iop.ProviderTunnelFrame{
|
||||
Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY,
|
||||
Body: tt.terminalFrame,
|
||||
}
|
||||
close(frames)
|
||||
|
||||
select {
|
||||
case runErr := <-runDone:
|
||||
if runErr != nil {
|
||||
t.Fatalf("runtime.Run: %v", runErr)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("runtime did not finish after large event evaluation")
|
||||
}
|
||||
if closeErr := runtime.CloseRequestResources(t.Context(), true); closeErr != nil {
|
||||
t.Fatalf("CloseRequestResources: %v", closeErr)
|
||||
}
|
||||
|
||||
status, headerCalls, body := writer.snapshot()
|
||||
if headerCalls != 1 {
|
||||
t.Fatalf("response-start commits = %d, want 1", headerCalls)
|
||||
}
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("HTTP status = %d, want %d", status, http.StatusOK)
|
||||
}
|
||||
terminalCommitted, terminalSuccess := sink.terminalStatus()
|
||||
if !terminalCommitted || !terminalSuccess {
|
||||
t.Fatalf("terminal status = (committed=%v, success=%v), want (true, true)", terminalCommitted, terminalSuccess)
|
||||
}
|
||||
wantBody := string(tt.semanticFrame) + string(tt.terminalFrame)
|
||||
if string(body) != wantBody {
|
||||
t.Fatalf("released body length = %d, want exact wire length %d", len(body), len(wantBody))
|
||||
}
|
||||
|
||||
for _, obs := range observations.Snapshot() {
|
||||
if obs.Kind() == streamgate.ObservationKindRecoveryDispatched {
|
||||
t.Fatalf("unexpected recovery dispatched observation: %+v", obs)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue