iop/apps/node/internal/node/sink_test.go

package node

import (
	"bytes"
	"context"
	"strings"
	"testing"
	"time"

	"google.golang.org/protobuf/proto"

	"iop/apps/node/internal/runtime"
	iop "iop/proto/gen/iop"
)

// mockSender is a minimal mock for protoSender.
type mockSender struct {
	sent       *iop.RunEvent
	sentEvents []*iop.RunEvent
}

func (m *mockSender) Send(msg proto.Message) error {
	if re, ok := msg.(*iop.RunEvent); ok {
		m.sent = re
		m.sentEvents = append(m.sentEvents, re)
	}
	return nil
}

func TestSessionSinkEmitIncludesNodeID(t *testing.T) {
	ms := &mockSender{}
	sink := &sessionSink{
		sess:       ms,
		nodeID:     "test-node-123",
		sessionID:  "session-abc",
		background: true,
	}

	event := runtime.RuntimeEvent{
		RunID:     "run-xyz",
		Type:      runtime.EventTypeStart,
		Timestamp: time.Now(),
	}

	err := sink.Emit(context.Background(), event)
	if err != nil {
		t.Fatalf("Emit failed: %v", err)
	}

	if ms.sent == nil {
		t.Fatal("no event sent")
	}

	if ms.sent.NodeId != "test-node-123" {
		t.Errorf("expected NodeId %q, got %q", "test-node-123", ms.sent.NodeId)
	}
	if ms.sent.SessionId != "session-abc" {
		t.Errorf("expected SessionId %q, got %q", "session-abc", ms.sent.SessionId)
	}
	if !ms.sent.Background {
		t.Error("expected Background true")
	}
}

func TestSessionSinkPrintEventStreamsDeltaImmediately(t *testing.T) {
	ms := &mockSender{}
	var out bytes.Buffer
	sink := &sessionSink{
		sess:      ms,
		out:       &out,
		nodeID:    "test-node-123",
		sessionID: "session-abc",
	}

	start := runtime.RuntimeEvent{
		RunID:     "run-xyz",
		Type:      runtime.EventTypeStart,
		Timestamp: time.Now(),
	}
	delta1 := runtime.RuntimeEvent{
		RunID:     "run-xyz",
		Type:      runtime.EventTypeDelta,
		Delta:     "Hello",
		Timestamp: time.Now(),
	}
	delta2 := runtime.RuntimeEvent{
		RunID:     "run-xyz",
		Type:      runtime.EventTypeDelta,
		Delta:     " world",
		Timestamp: time.Now(),
	}
	complete := runtime.RuntimeEvent{
		RunID:     "run-xyz",
		Type:      runtime.EventTypeComplete,
		Message:   "done",
		Timestamp: time.Now(),
	}

	if err := sink.Emit(context.Background(), start); err != nil {
		t.Fatalf("Emit start failed: %v", err)
	}
	if err := sink.Emit(context.Background(), delta1); err != nil {
		t.Fatalf("Emit delta1 failed: %v", err)
	}
	got := out.String()
	if !strings.Contains(got, "[node-message] Hello") {
		t.Fatalf("expected first delta to be printed immediately, got %q", got)
	}

	if err := sink.Emit(context.Background(), delta2); err != nil {
		t.Fatalf("Emit delta2 failed: %v", err)
	}
	got = out.String()
	if !strings.Contains(got, "[node-message] Hello world") {
		t.Fatalf("expected second delta to append to stream, got %q", got)
	}

	if err := sink.Emit(context.Background(), complete); err != nil {
		t.Fatalf("Emit complete failed: %v", err)
	}
	got = out.String()
	if !strings.Contains(got, "[node-event] complete run_id=run-xyz detail=\"done\"") {
		t.Fatalf("expected complete event to be printed, got %q", got)
	}
	if strings.Count(got, "[node-message]") != 1 {
		t.Fatalf("expected node-message prefix once, got %q", got)
	}
}

func TestTerminalDeferringSinkFlushesTerminalEvents(t *testing.T) {
	ms := &mockSender{}
	var out bytes.Buffer
	inner := &sessionSink{
		sess:      ms,
		out:       &out,
		nodeID:    "test-node-123",
		sessionID: "session-abc",
	}
	sink := &terminalDeferringSink{inner: inner}

	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     "run-terminal",
		Type:      runtime.EventTypeStart,
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit start failed: %v", err)
	}
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     "run-terminal",
		Type:      runtime.EventTypeComplete,
		Message:   "done",
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit complete failed: %v", err)
	}
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     "run-terminal",
		Type:      runtime.EventTypeDelta,
		Delta:     "late",
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit late delta failed: %v", err)
	}

	if len(ms.sentEvents) != 1 {
		t.Fatalf("expected only start to be sent before flush, got %d events", len(ms.sentEvents))
	}
	if got := ms.sentEvents[0].GetType(); got != string(runtime.EventTypeStart) {
		t.Fatalf("first event type = %q, want start", got)
	}
	if strings.Contains(out.String(), "complete run_id=run-terminal") {
		t.Fatalf("complete was printed before flush: %q", out.String())
	}

	if err := sink.Flush(context.Background()); err != nil {
		t.Fatalf("Flush failed: %v", err)
	}
	if len(ms.sentEvents) != 3 {
		t.Fatalf("expected start, complete, late delta after flush, got %d events", len(ms.sentEvents))
	}
	if got := ms.sentEvents[1].GetType(); got != string(runtime.EventTypeComplete) {
		t.Fatalf("second event type = %q, want complete", got)
	}
	if got := ms.sentEvents[2].GetType(); got != string(runtime.EventTypeDelta) {
		t.Fatalf("third event type = %q, want delta", got)
	}
	if !strings.Contains(out.String(), "[node-event] complete run_id=run-terminal detail=\"done\"") {
		t.Fatalf("complete was not printed after flush: %q", out.String())
	}
}

func TestTerminalDeferringSinkRecordsTerminal(t *testing.T) {
	ms := &mockSender{}
	inner := &sessionSink{
		sess:      ms,
		nodeID:    "test-node",
		sessionID: "session-x",
	}
	sink := &terminalDeferringSink{inner: inner}

	// Emit a complete terminal event.
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     "run-1",
		Type:      runtime.EventTypeComplete,
		Message:   "ok",
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit complete: %v", err)
	}
	if !sink.hasTerminalObserved() {
		t.Fatal("expected terminalObserved to be true after emitting a terminal event")
	}

	// Emit a delta (should not change terminalObserved).
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     "run-1",
		Type:      runtime.EventTypeDelta,
		Delta:     "x",
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit delta: %v", err)
	}
	if !sink.hasTerminalObserved() {
		t.Fatal("expected terminalObserved to remain true")
	}
}

func TestTerminalDeferringSinkNoTerminalNotMarked(t *testing.T) {
	ms := &mockSender{}
	inner := &sessionSink{
		sess:      ms,
		nodeID:    "test-node",
		sessionID: "session-x",
	}
	sink := &terminalDeferringSink{inner: inner}

	// Only emit start and delta (no terminal event).
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     "run-2",
		Type:      runtime.EventTypeStart,
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit start: %v", err)
	}
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     "run-2",
		Type:      runtime.EventTypeDelta,
		Delta:     "data",
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit delta: %v", err)
	}
	if sink.hasTerminalObserved() {
		t.Fatal("expected terminalObserved to be false when no terminal event emitted")
	}
}

func TestTerminalDeferringSinkDoesNotDuplicateAdapterTerminalEvent(t *testing.T) {
	ms := &mockSender{}
	inner := &sessionSink{
		sess:      ms,
		nodeID:    "test-node",
		sessionID: "session-x",
	}
	sink := &terminalDeferringSink{inner: inner}

	// Adapter emits start then complete terminal event.
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     "run-3",
		Type:      runtime.EventTypeStart,
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit start: %v", err)
	}
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     "run-3",
		Type:      runtime.EventTypeComplete,
		Message:   "adapter done",
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit complete: %v", err)
	}

	// Flush should send both events.
	if err := sink.Flush(context.Background()); err != nil {
		t.Fatalf("Flush: %v", err)
	}
	if len(ms.sentEvents) != 2 {
		t.Fatalf("expected 2 events (start + complete), got %d", len(ms.sentEvents))
	}
	if ms.sentEvents[1].GetType() != string(runtime.EventTypeComplete) {
		t.Fatalf("second event type = %q, want complete", ms.sentEvents[1].GetType())
	}
	// Verify the adapter's own terminal event message is preserved (no synthetic replacement).
	if ms.sentEvents[1].Message != "adapter done" {
		t.Fatalf("expected Message=\"adapter done\", got %q", ms.sentEvents[1].Message)
	}
}

func TestTerminalDeferringSinkSynthesizedTerminalNotDuplicated(t *testing.T) {
	ms := &mockSender{}
	inner := &sessionSink{
		sess:      ms,
		nodeID:    "test-node",
		sessionID: "session-x",
	}
	sink := &terminalDeferringSink{inner: inner}

	// Adapter emits start and delta, but NO terminal event.
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     "run-4",
		Type:      runtime.EventTypeStart,
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit start: %v", err)
	}
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     "run-4",
		Type:      runtime.EventTypeDelta,
		Delta:     "data",
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit delta: %v", err)
	}

	// Verify no terminal was observed.
	if sink.hasTerminalObserved() {
		t.Fatal("expected no terminal observed")
	}

	// Synthesize a terminal event (simulating synthAndEmitTerminal).
	synthEvent := runtime.RuntimeEvent{
		RunID:     "run-4",
		Type:      runtime.EventTypeComplete,
		Message:   "adapter completed without terminal event",
		Timestamp: time.Now(),
	}
	if err := sink.Emit(context.Background(), synthEvent); err != nil {
		t.Fatalf("Emit synthesized complete: %v", err)
	}

	// Flush should send all events including the synthesized one.
	if err := sink.Flush(context.Background()); err != nil {
		t.Fatalf("Flush: %v", err)
	}
	if len(ms.sentEvents) != 3 {
		t.Fatalf("expected 3 events (start, delta, complete), got %d", len(ms.sentEvents))
	}
	if ms.sentEvents[2].GetType() != string(runtime.EventTypeComplete) {
		t.Fatalf("third event type = %q, want complete", ms.sentEvents[2].GetType())
	}
	// Verify the synthesized event's message.
	if ms.sentEvents[2].Message != "adapter completed without terminal event" {
		t.Fatalf("expected synthesized Message=\"adapter completed without terminal event\", got %q", ms.sentEvents[2].Message)
	}
	// terminalObserved should now be true.
	if !sink.hasTerminalObserved() {
		t.Fatal("expected terminalObserved to be true after flush")
	}
}

func TestSessionSinkPrintEventPreservesReasoningDelta(t *testing.T) {
	ms := &mockSender{}
	var out bytes.Buffer
	sink := &sessionSink{
		sess:      ms,
		out:       &out,
		nodeID:    "test-node",
		sessionID: "session-x",
	}

	runID := "run-reasoning"

	// Emit start.
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     runID,
		Type:      runtime.EventTypeStart,
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit start: %v", err)
	}

	// Emit reasoning delta with multiline content.
	reasoning := runtime.RuntimeEvent{
		RunID:     runID,
		Type:      runtime.EventTypeReasoningDelta,
		Delta:     "사용자가 인사했습니다.\n😀\n",
		Timestamp: time.Now(),
	}
	if err := sink.Emit(context.Background(), reasoning); err != nil {
		t.Fatalf("Emit reasoning_delta: %v", err)
	}

	// Emit complete.
	if err := sink.Emit(context.Background(), runtime.RuntimeEvent{
		RunID:     runID,
		Type:      runtime.EventTypeComplete,
		Message:   "done",
		Timestamp: time.Now(),
	}); err != nil {
		t.Fatalf("Emit complete: %v", err)
	}

	got := out.String()

	// Reasoning delta should be preserved with [node-reasoning] prefix.
	if !strings.Contains(got, "[node-reasoning] 사용자가 인사했습니다.") {
		t.Fatalf("expected [node-reasoning] line 1, got %q", got)
	}
	if !strings.Contains(got, "[node-reasoning] 😀") {
		if !strings.Contains(got, "[node-reasoning] ") {
			t.Fatalf("expected [node-reasoning] prefix in stdout, got %q", got)
		}
	}

	// stdout should NOT contain the raw reasoning_delta fallback.
	if strings.Contains(got, "reasoning_delta run_id=") && strings.Contains(got, "detail=\"\"") {
		t.Fatalf("reasoning_delta should not fall through to default case, got %q", got)
	}

	// [node-message] should not appear for reasoning-only output.
	if strings.Contains(got, "[node-message]") {
		t.Fatalf("[node-message] should not appear for reasoning-only output, got %q", got)
	}

	// Verify proto event preserves reasoning_delta type and delta content.
	if len(ms.sentEvents) != 3 {
		t.Fatalf("expected 3 proto events, got %d", len(ms.sentEvents))
	}
	if ms.sentEvents[1].GetType() != "reasoning_delta" {
		t.Fatalf("proto event type = %q, want reasoning_delta", ms.sentEvents[1].GetType())
	}
	if ms.sentEvents[1].Delta != "사용자가 인사했습니다.\n😀\n" {
		t.Fatalf("proto event Delta = %q, expected multiline reasoning text", ms.sentEvents[1].Delta)
	}
}