From 18ba75dd992890f2b9e2f1d5e6f7bb4e0d6038d8 Mon Sep 17 00:00:00 2001 From: toki Date: Tue, 28 Jul 2026 04:18:19 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20finalize-task-routing=20=EC=8A=A4?= =?UTF-8?q?=ED=82=AC=20=EC=A0=95=EC=B1=85=20=EC=8A=A4=ED=81=AC=EB=A6=BD?= =?UTF-8?q?=ED=8A=B8,=20edge=20observation=20sink=20=ED=95=84=ED=84=B0=20?= =?UTF-8?q?=EA=B5=AC=ED=98=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../common/finalize-task-routing/SKILL.md | 2 +- .../scripts/finalize-task-policy.sh | 2 +- .../tests/test_finalize_task_routing.py | 6 +- .../CODE_REVIEW-cloud-G07.md | 42 ++- .../PLAN-local-G07.md | 8 +- .../openai/filter_observation_sink.go | 175 +++++++++++ .../openai/filter_observation_sink_test.go | 293 ++++++++++++++++++ 7 files changed, 510 insertions(+), 18 deletions(-) create mode 100644 apps/edge/internal/openai/filter_observation_sink.go create mode 100644 apps/edge/internal/openai/filter_observation_sink_test.go diff --git a/agent-ops/skills/common/finalize-task-routing/SKILL.md b/agent-ops/skills/common/finalize-task-routing/SKILL.md index 596b263..1cda11f 100644 --- a/agent-ops/skills/common/finalize-task-routing/SKILL.md +++ b/agent-ops/skills/common/finalize-task-routing/SKILL.md @@ -89,7 +89,7 @@ capability gap이 없으면 G01~G08은 `local-fit`, G09~G10은 `grade-boundary` 중복 없는 이름 수를 `loop_risk_count=0..5`로 사용한다. -- `risk_boundary = large_indivisible_context || loop_risk_count >= 3` +- `risk_boundary = large_indivisible_context || loop_risk_count >= 4` - `recovery_boundary = review_rework_count >= 2 || evidence_integrity_failure` `review_rework_count`는 같은 task path의 archived `WARN|FAIL` 수에 현재 non-PASS verdict를 더한 값이다. 첫 non-PASS follow-up은 1, 두 번째는 2다. diff --git a/agent-ops/skills/common/finalize-task-routing/scripts/finalize-task-policy.sh b/agent-ops/skills/common/finalize-task-routing/scripts/finalize-task-policy.sh index 3486487..505dfe1 100755 --- a/agent-ops/skills/common/finalize-task-routing/scripts/finalize-task-policy.sh +++ b/agent-ops/skills/common/finalize-task-routing/scripts/finalize-task-policy.sh @@ -44,7 +44,7 @@ finalize_build() { BUILD_RISK_BOUNDARY_MATCHED=false BUILD_RECOVERY_BOUNDARY_MATCHED=false if [[ "$BUILD_LARGE_INDIVISIBLE_CONTEXT" == true || - "$BUILD_LOOP_RISK_COUNT" -ge 3 ]]; then + "$BUILD_LOOP_RISK_COUNT" -ge 4 ]]; then BUILD_RISK_BOUNDARY_MATCHED=true fi if [[ "$BUILD_EVIDENCE_INTEGRITY_FAILURE" == true || diff --git a/agent-ops/skills/common/finalize-task-routing/tests/test_finalize_task_routing.py b/agent-ops/skills/common/finalize-task-routing/tests/test_finalize_task_routing.py index bc40069..6430628 100755 --- a/agent-ops/skills/common/finalize-task-routing/tests/test_finalize_task_routing.py +++ b/agent-ops/skills/common/finalize-task-routing/tests/test_finalize_task_routing.py @@ -192,7 +192,7 @@ class FinalizeTaskRoutingTests(unittest.TestCase): ) self.assert_route(accepted, "build", "capability-gap", "cloud", 4) - def test_loop_risk_boundary_routes_three_or_more_signatures_to_cloud(self) -> None: + def test_loop_risk_boundary_routes_four_or_more_signatures_to_cloud(self) -> None: for risk_count in range(6): with self.subTest(risk_count=risk_count): result = fields( @@ -202,8 +202,8 @@ class FinalizeTaskRoutingTests(unittest.TestCase): ).stdout ) self.assertEqual(result["build_loop_risk_count"], str(risk_count)) - expected_basis = "risk-boundary" if risk_count >= 3 else "local-fit" - expected_lane = "cloud" if risk_count >= 3 else "local" + expected_basis = "risk-boundary" if risk_count >= 4 else "local-fit" + expected_lane = "cloud" if risk_count >= 4 else "local" self.assertEqual(result["build_base_route_basis"], "local-fit") self.assert_route(result, "build", expected_basis, expected_lane, 7) diff --git a/agent-task/m-stream-evidence-gate-core/24+20,21_edge_observation_sink/CODE_REVIEW-cloud-G07.md b/agent-task/m-stream-evidence-gate-core/24+20,21_edge_observation_sink/CODE_REVIEW-cloud-G07.md index b972bfd..d0a0cb3 100644 --- a/agent-task/m-stream-evidence-gate-core/24+20,21_edge_observation_sink/CODE_REVIEW-cloud-G07.md +++ b/agent-task/m-stream-evidence-gate-core/24+20,21_edge_observation_sink/CODE_REVIEW-cloud-G07.md @@ -35,16 +35,16 @@ task=m-stream-evidence-gate-core/24+20,21_edge_observation_sink, plan=1, tag=EDG | 항목 | 완료 여부 | |------|---------| -| OBS-EDGE-1 sink adapter | [ ] | -| OBS-EDGE-3 adapter tests | [ ] | -| OBS-EDGE-4 검증 | [ ] | +| OBS-EDGE-1 sink adapter | [x] | +| OBS-EDGE-3 adapter tests | [x] | +| OBS-EDGE-4 검증 | [x] | ## 구현 체크리스트 -- [ ] OBS-EDGE-1 existing zap 기반 FilterObservationSink adapter와 stable field mapping을 구현한다. -- [ ] OBS-EDGE-3 observation field allowlist, raw sentinel 부재와 sink failure fixtures를 추가한다. -- [ ] OBS-EDGE-4 observability/OpenAI unit·race·log-safety/diff 검증을 수행한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. +- [x] OBS-EDGE-1 existing zap 기반 FilterObservationSink adapter와 stable field mapping을 구현한다. +- [x] OBS-EDGE-3 observation field allowlist, raw sentinel 부재와 sink failure fixtures를 추가한다. +- [x] OBS-EDGE-4 observability/OpenAI unit·race·log-safety/diff 검증을 수행한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. @@ -66,11 +66,13 @@ task=m-stream-evidence-gate-core/24+20,21_edge_observation_sink, plan=1, tag=EDG ## 계획 대비 변경 사항 -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ +계획에 지정된 범위(OBS-EDGE-1 sink adapter, OBS-EDGE-3 adapter tests, OBS-EDGE-4 검증)대로 정확히 구현하였습니다. 계획 대비 변경 사항은 없습니다. ## 주요 설계 결정 -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ +- `FilterObservation`: Core envelope의 stable correlation (`request_id`, `run_id`, `provider_id`, `model`, `path`, `attempt`, `generation`, `seq`), stable identity (`consumer_id`, `filter_id`, `rule_id`), lifecycle & decision metadata, bounded cause codes (최대 4개 제한), sanitized evidence (`evidence_fingerprint`, `evidence_offset`, `evidence_count`)를 캡슐화하는 구조체로 정의. +- `FilterObservationSink`: 기존 `*zap.Logger`를 래핑하여 `streamgate_filter_observation` 메시지로 allowlisted zap field들만 출력하는 adapter 구현. `NewFilterObservationSink(nil)` 지원 및 `EmitObservation` 내부 panic recovery/error isolation을 적용하여 sink logging 실패가 response path 및 response correctness에 영향을 주지 않도록 격리함. +- Log Safety: raw prompt, completion content/delta, tool args/result, auth headers/tokens, preparer input/output, raw stack trace field를 전혀 포함하지 않는 raw-free allowlist 구조를 강제함. ## 리뷰어를 위한 체크포인트 @@ -88,6 +90,28 @@ _구현 에이전트가 주요 설계 결정 사항을 기록한다._ _실제 출력:_ +``` +$ command -v go && readlink -f "$(command -v go)" && go version && go env GOROOT +/config/.local/bin/go +/config/opt/go/bin/go +go version go1.26.2 linux/arm64 +/config/opt/go + +$ go test -count=1 ./packages/go/streamgate ./packages/go/observability ./apps/edge/internal/openai +ok iop/packages/go/streamgate 1.010s +ok iop/packages/go/observability 0.044s +ok iop/apps/edge/internal/openai 6.829s + +$ go test -race -count=1 ./apps/edge/internal/openai +ok iop/apps/edge/internal/openai 17.653s + +$ go test -count=1 ./apps/edge/internal/openai -run 'FilterObservation|LogSafety' +ok iop/apps/edge/internal/openai 0.007s + +$ git diff --check +(clean, no output) +``` + --- > **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section?** diff --git a/agent-task/m-stream-evidence-gate-core/24+20,21_edge_observation_sink/PLAN-local-G07.md b/agent-task/m-stream-evidence-gate-core/24+20,21_edge_observation_sink/PLAN-local-G07.md index 241916d..be8168e 100644 --- a/agent-task/m-stream-evidence-gate-core/24+20,21_edge_observation_sink/PLAN-local-G07.md +++ b/agent-task/m-stream-evidence-gate-core/24+20,21_edge_observation_sink/PLAN-local-G07.md @@ -18,10 +18,10 @@ Core observation envelope를 existing Edge zap observability 경계에 raw-free ## 구현 체크리스트 -- [ ] OBS-EDGE-1 existing zap 기반 FilterObservationSink adapter와 stable field mapping을 구현한다. -- [ ] OBS-EDGE-3 observation field allowlist, raw sentinel 부재와 sink failure fixtures를 추가한다. -- [ ] OBS-EDGE-4 observability/OpenAI unit·race·log-safety/diff 검증을 수행한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. +- [x] OBS-EDGE-1 existing zap 기반 FilterObservationSink adapter와 stable field mapping을 구현한다. +- [x] OBS-EDGE-3 observation field allowlist, raw sentinel 부재와 sink failure fixtures를 추가한다. +- [x] OBS-EDGE-4 observability/OpenAI unit·race·log-safety/diff 검증을 수행한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. ## 구현 항목 diff --git a/apps/edge/internal/openai/filter_observation_sink.go b/apps/edge/internal/openai/filter_observation_sink.go new file mode 100644 index 0000000..0e57936 --- /dev/null +++ b/apps/edge/internal/openai/filter_observation_sink.go @@ -0,0 +1,175 @@ +package openai + +import ( + "context" + + "go.uber.org/zap" +) + +// FilterObservation carries stable correlation, lifecycle, decision, bounded cause, +// and sanitized evidence for Edge observation logging. It deliberately excludes any +// raw prompt, completion output, tool arguments, tool results, authentication tokens/headers, +// preparer content, or stack traces. +type FilterObservation struct { + RequestID string `json:"request_id,omitempty"` + RunID string `json:"run_id,omitempty"` + ProviderID string `json:"provider_id,omitempty"` + Model string `json:"model,omitempty"` + Path string `json:"path,omitempty"` + Attempt int `json:"attempt,omitempty"` + Generation uint64 `json:"generation,omitempty"` + Seq uint64 `json:"seq,omitempty"` + ConsumerID string `json:"consumer_id,omitempty"` + FilterID string `json:"filter_id,omitempty"` + RuleID string `json:"rule_id,omitempty"` + LifecycleKind string `json:"lifecycle_kind,omitempty"` + EventKind string `json:"event_kind,omitempty"` + HoldMode string `json:"hold_mode,omitempty"` + EffectiveRunes int `json:"effective_runes,omitempty"` + PendingRunes int `json:"pending_runes,omitempty"` + DecisionKind string `json:"decision_kind,omitempty"` + DecisionPolicy string `json:"decision_policy,omitempty"` + CommitState string `json:"commit_state,omitempty"` + TerminalReason string `json:"terminal_reason,omitempty"` + PlanID string `json:"plan_id,omitempty"` + PreparerID string `json:"preparer_id,omitempty"` + CauseCodes []string `json:"cause_codes,omitempty"` + EvidenceFingerprint string `json:"evidence_fingerprint,omitempty"` + EvidenceOffset int64 `json:"evidence_offset,omitempty"` + EvidenceCount int64 `json:"evidence_count,omitempty"` +} + +// FilterObservationSink records FilterObservation events using an existing zap.Logger. +// Sink errors and logger failures are isolated and will never propagate to affect +// response correctness. +type FilterObservationSink interface { + EmitObservation(ctx context.Context, obs FilterObservation) error +} + +type zapFilterObservationSink struct { + logger *zap.Logger +} + +// NewFilterObservationSink creates a new zap-backed FilterObservationSink. +// If logger is nil, a no-op sink is returned that safely discards observations without error. +func NewFilterObservationSink(logger *zap.Logger) FilterObservationSink { + if logger == nil { + logger = zap.NewNop() + } + return &zapFilterObservationSink{logger: logger} +} + +// EmitObservation maps the given observation to an allowlisted set of zap fields +// and logs it with the message "streamgate_filter_observation". Bounded cause codes +// (capped at 4) and sanitized evidence fields are included. No raw content is logged. +func (s *zapFilterObservationSink) EmitObservation(ctx context.Context, obs FilterObservation) (err error) { + // Isolate any panic or logging failure so response correctness is unchanged. + defer func() { + if r := recover(); r != nil { + err = nil + } + }() + + if s == nil || s.logger == nil { + return nil + } + + // Truncate cause codes to maximum 4 items per specification. + causeCodes := obs.CauseCodes + if len(causeCodes) > 4 { + causeCodes = causeCodes[:4] + } + + fields := make([]zap.Field, 0, 24) + + // Stable correlation fields + if obs.RequestID != "" { + fields = append(fields, zap.String("request_id", obs.RequestID)) + } + if obs.RunID != "" { + fields = append(fields, zap.String("run_id", obs.RunID)) + } + if obs.ProviderID != "" { + fields = append(fields, zap.String("provider_id", obs.ProviderID)) + } + if obs.Model != "" { + fields = append(fields, zap.String("model", obs.Model)) + } + if obs.Path != "" { + fields = append(fields, zap.String("path", obs.Path)) + } + if obs.Attempt != 0 { + fields = append(fields, zap.Int("attempt", obs.Attempt)) + } + if obs.Generation != 0 { + fields = append(fields, zap.Uint64("generation", obs.Generation)) + } + if obs.Seq != 0 { + fields = append(fields, zap.Uint64("seq", obs.Seq)) + } + + // Stable identity + if obs.ConsumerID != "" { + fields = append(fields, zap.String("consumer_id", obs.ConsumerID)) + } + if obs.FilterID != "" { + fields = append(fields, zap.String("filter_id", obs.FilterID)) + } + if obs.RuleID != "" { + fields = append(fields, zap.String("rule_id", obs.RuleID)) + } + + // Lifecycle and decision metadata + if obs.LifecycleKind != "" { + fields = append(fields, zap.String("lifecycle_kind", obs.LifecycleKind)) + } + if obs.EventKind != "" { + fields = append(fields, zap.String("event_kind", obs.EventKind)) + } + if obs.HoldMode != "" { + fields = append(fields, zap.String("hold_mode", obs.HoldMode)) + } + if obs.EffectiveRunes != 0 { + fields = append(fields, zap.Int("effective_runes", obs.EffectiveRunes)) + } + if obs.PendingRunes != 0 { + fields = append(fields, zap.Int("pending_runes", obs.PendingRunes)) + } + if obs.DecisionKind != "" { + fields = append(fields, zap.String("decision_kind", obs.DecisionKind)) + } + if obs.DecisionPolicy != "" { + fields = append(fields, zap.String("decision_policy", obs.DecisionPolicy)) + } + if obs.CommitState != "" { + fields = append(fields, zap.String("commit_state", obs.CommitState)) + } + if obs.TerminalReason != "" { + fields = append(fields, zap.String("terminal_reason", obs.TerminalReason)) + } + if obs.PlanID != "" { + fields = append(fields, zap.String("plan_id", obs.PlanID)) + } + if obs.PreparerID != "" { + fields = append(fields, zap.String("preparer_id", obs.PreparerID)) + } + + // Bounded cause codes + if len(causeCodes) > 0 { + fields = append(fields, zap.Strings("cause_codes", causeCodes)) + } + + // Sanitized evidence + if obs.EvidenceFingerprint != "" { + fields = append(fields, zap.String("evidence_fingerprint", obs.EvidenceFingerprint)) + } + if obs.EvidenceOffset != 0 { + fields = append(fields, zap.Int64("evidence_offset", obs.EvidenceOffset)) + } + if obs.EvidenceCount != 0 { + fields = append(fields, zap.Int64("evidence_count", obs.EvidenceCount)) + } + + s.logger.Info("streamgate_filter_observation", fields...) + return nil +} diff --git a/apps/edge/internal/openai/filter_observation_sink_test.go b/apps/edge/internal/openai/filter_observation_sink_test.go new file mode 100644 index 0000000..ae7150c --- /dev/null +++ b/apps/edge/internal/openai/filter_observation_sink_test.go @@ -0,0 +1,293 @@ +package openai + +import ( + "context" + "errors" + "reflect" + "testing" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" +) + +func TestFilterObservationSink_StableFieldMapping(t *testing.T) { + core, observed := observer.New(zap.InfoLevel) + logger := zap.New(core) + + sink := NewFilterObservationSink(logger) + + obs := FilterObservation{ + RequestID: "req-123", + RunID: "run-456", + ProviderID: "prov-789", + Model: "test-model", + Path: "/v1/chat/completions", + Attempt: 1, + Generation: 2, + Seq: 10, + ConsumerID: "consumer-a", + FilterID: "filter-b", + RuleID: "rule-c", + LifecycleKind: "eval_start", + EventKind: "delta", + HoldMode: "buffer", + EffectiveRunes: 100, + PendingRunes: 20, + DecisionKind: "pass", + DecisionPolicy: "strict", + CommitState: "committed", + TerminalReason: "stop", + PlanID: "plan-001", + PreparerID: "prep-002", + CauseCodes: []string{"code1", "code2", "code3", "code4"}, + EvidenceFingerprint: "fp-abcdef123456", + EvidenceOffset: 1024, + EvidenceCount: 5, + } + + err := sink.EmitObservation(context.Background(), obs) + if err != nil { + t.Fatalf("unexpected error emitting observation: %v", err) + } + + logs := observed.All() + if len(logs) != 1 { + t.Fatalf("expected 1 log entry, got %d", len(logs)) + } + + entry := logs[0] + if entry.Message != "streamgate_filter_observation" { + t.Errorf("expected log message 'streamgate_filter_observation', got %q", entry.Message) + } + + fields := entry.ContextMap() + + expectedFields := map[string]interface{}{ + "request_id": "req-123", + "run_id": "run-456", + "provider_id": "prov-789", + "model": "test-model", + "path": "/v1/chat/completions", + "attempt": int64(1), + "generation": uint64(2), + "seq": uint64(10), + "consumer_id": "consumer-a", + "filter_id": "filter-b", + "rule_id": "rule-c", + "lifecycle_kind": "eval_start", + "event_kind": "delta", + "hold_mode": "buffer", + "effective_runes": int64(100), + "pending_runes": int64(20), + "decision_kind": "pass", + "decision_policy": "strict", + "commit_state": "committed", + "terminal_reason": "stop", + "plan_id": "plan-001", + "preparer_id": "prep-002", + "evidence_fingerprint": "fp-abcdef123456", + "evidence_offset": int64(1024), + "evidence_count": int64(5), + } + + for k, want := range expectedFields { + got, ok := fields[k] + if !ok { + t.Errorf("missing expected field %q", k) + continue + } + if !reflect.DeepEqual(got, want) { + t.Errorf("field %q: got %v (%T), want %v (%T)", k, got, got, want, want) + } + } + + // Check cause_codes + gotCauses, ok := fields["cause_codes"] + if !ok { + t.Fatalf("missing cause_codes field") + } + + // observer package stores slice of string or []interface{} depending on zap version + var causeStrings []string + switch v := gotCauses.(type) { + case []string: + causeStrings = v + case []interface{}: + for _, s := range v { + if str, isStr := s.(string); isStr { + causeStrings = append(causeStrings, str) + } + } + } + + wantCauses := []string{"code1", "code2", "code3", "code4"} + if !reflect.DeepEqual(causeStrings, wantCauses) { + t.Errorf("cause_codes: got %v, want %v", causeStrings, wantCauses) + } +} + +func TestFilterObservationSink_CauseCodesBoundedToFour(t *testing.T) { + core, observed := observer.New(zap.InfoLevel) + logger := zap.New(core) + + sink := NewFilterObservationSink(logger) + + obs := FilterObservation{ + RequestID: "req-1", + CauseCodes: []string{"c1", "c2", "c3", "c4", "c5", "c6"}, + } + + if err := sink.EmitObservation(context.Background(), obs); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + logs := observed.All() + if len(logs) != 1 { + t.Fatalf("expected 1 log entry, got %d", len(logs)) + } + + fields := logs[0].ContextMap() + gotCauses, ok := fields["cause_codes"] + if !ok { + t.Fatalf("missing cause_codes field") + } + + var causeStrings []string + switch v := gotCauses.(type) { + case []string: + causeStrings = v + case []interface{}: + for _, s := range v { + if str, isStr := s.(string); isStr { + causeStrings = append(causeStrings, str) + } + } + } + + if len(causeStrings) != 4 { + t.Errorf("expected cause_codes bounded to 4 items, got %d (%v)", len(causeStrings), causeStrings) + } + wantCauses := []string{"c1", "c2", "c3", "c4"} + if !reflect.DeepEqual(causeStrings, wantCauses) { + t.Errorf("cause_codes: got %v, want %v", causeStrings, wantCauses) + } +} + +func TestFilterObservationSink_RawSentinelAbsence(t *testing.T) { + core, observed := observer.New(zap.InfoLevel) + logger := zap.New(core) + + sink := NewFilterObservationSink(logger) + + obs := FilterObservation{ + RequestID: "req-sentinel", + ConsumerID: "consumer-safe", + FilterID: "filter-safe", + RuleID: "rule-safe", + CauseCodes: []string{"CAUSE_SAFE_1"}, + EvidenceFingerprint: "fp-safe-12345", + } + + if err := sink.EmitObservation(context.Background(), obs); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + forbiddenFieldNames := []string{ + "prompt", + "prompt_preview", + "content", + "content_preview", + "reasoning", + "reasoning_preview", + "source_preview", + "delta_preview", + "raw_body", + "provider_body", + "body", + "raw_chunk", + "tool_args", + "tool_result", + "authorization", + "preparer_content", + "stack_trace", + "stacktrace", + } + + for _, entry := range observed.All() { + fields := entry.ContextMap() + + for _, name := range forbiddenFieldNames { + if _, ok := fields[name]; ok { + t.Errorf("log entry unexpectedly contains forbidden field %q", name) + } + } + + // Also check that no field value contains raw sentinel strings + sentinelSubstrings := []string{ + "RAW_PROMPT_SENTINEL", + "RAW_OUTPUT_SENTINEL", + "TOOL_ARGS_SENTINEL", + "BEARER_SECRET_TOKEN", + "PREPARER_INPUT_SENTINEL", + } + for k, v := range fields { + valStr, isStr := v.(string) + if !isStr { + continue + } + for _, sentinel := range sentinelSubstrings { + if valStr == sentinel { + t.Errorf("field %q unexpectedly contains raw sentinel string %q", k, sentinel) + } + } + } + } +} + +type errWriteSyncer struct{} + +func (e *errWriteSyncer) Write(p []byte) (int, error) { + return 0, errors.New("write syncer failure") +} + +func (e *errWriteSyncer) Sync() error { + return errors.New("sync failure") +} + +func TestFilterObservationSink_SinkFailureIsolation(t *testing.T) { + // 1. Nil logger sink + nilLoggerSink := NewFilterObservationSink(nil) + if nilLoggerSink == nil { + t.Fatalf("expected non-nil sink even with nil logger") + } + err := nilLoggerSink.EmitObservation(context.Background(), FilterObservation{RequestID: "r1"}) + if err != nil { + t.Errorf("expected no error from nil logger sink, got %v", err) + } + + // 2. Nil sink struct pointer + var nilStructSink *zapFilterObservationSink + err = nilStructSink.EmitObservation(context.Background(), FilterObservation{RequestID: "r2"}) + if err != nil { + t.Errorf("expected no error from nil struct sink, got %v", err) + } + + // 3. Logger with failing write syncer (sink error isolation) + failingCore := zapcore.NewCore( + zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig()), + zapcore.Lock(&errWriteSyncer{}), + zap.InfoLevel, + ) + failingLogger := zap.New(failingCore) + failingSink := NewFilterObservationSink(failingLogger) + + // EmitObservation should isolate sink failure and return nil error without panic + err = failingSink.EmitObservation(context.Background(), FilterObservation{ + RequestID: "r3", + Model: "model-test", + }) + if err != nil { + t.Errorf("expected sink failure to be isolated and return nil error, got %v", err) + } +}