package node

import (
	"context"
	"fmt"
	"io"
	"net"
	"sync"
	"testing"
	"time"

	toki "git.toki-labs.com/toki/proto-socket/go"
	"go.uber.org/zap"

	"google.golang.org/protobuf/proto"

	"iop/apps/node/internal/store"
	"iop/apps/node/internal/transport"
	runtime "iop/packages/go/execution"
	iop "iop/proto/gen/iop"
)

type manualAttemptTimer struct {
	mu          sync.Mutex
	ch          chan time.Time
	now         func() time.Time
	advanceTo   func(time.Time)
	durations   []time.Duration
	scheduled   time.Time
	stopped     bool
	fired       bool
	beforeReset func()
}

func newManualAttemptTimer(d time.Duration, now func() time.Time, advanceTo func(time.Time)) *manualAttemptTimer {
	scheduled := now().Add(d)
	return &manualAttemptTimer{ch: make(chan time.Time, 1), now: now, advanceTo: advanceTo, durations: []time.Duration{d}, scheduled: scheduled}
}
func (t *manualAttemptTimer) C() <-chan time.Time { return t.ch }
func (t *manualAttemptTimer) Stop() bool {
	t.mu.Lock()
	defer t.mu.Unlock()
	wasActive := !t.stopped && !t.fired
	t.stopped = true
	return wasActive
}
func (t *manualAttemptTimer) Reset(d time.Duration) bool {
	t.mu.Lock()
	beforeReset := t.beforeReset
	t.mu.Unlock()
	if beforeReset != nil {
		beforeReset()
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	wasStopped := t.stopped
	t.stopped = false
	t.fired = false
	t.durations = append(t.durations, d)
	t.scheduled = t.now().Add(d)
	return wasStopped
}
func (t *manualAttemptTimer) fire() {
	t.mu.Lock()
	stopped, fired, scheduled := t.stopped, t.fired, t.scheduled
	if !stopped && !fired {
		t.fired = true
	}
	t.mu.Unlock()
	if !stopped && !fired {
		t.advanceTo(scheduled)
		t.ch <- scheduled
	}
}
func (t *manualAttemptTimer) fireStaleArmDuringReset() {
	t.mu.Lock()
	scheduled := t.scheduled
	t.mu.Unlock()
	t.advanceTo(scheduled)
	t.ch <- scheduled
}
func (t *manualAttemptTimer) snapshot() ([]time.Duration, bool) {
	t.mu.Lock()
	defer t.mu.Unlock()
	return append([]time.Duration(nil), t.durations...), t.stopped
}

type manualAttemptClock struct {
	mu                sync.Mutex
	timers            []*manualAttemptTimer
	created           chan struct{}
	now               time.Time
	beforeTimerReturn func(*manualAttemptTimer)
}

func newManualAttemptClock() *manualAttemptClock {
	return &manualAttemptClock{created: make(chan struct{}, 16), now: time.Unix(0, 0)}
}

// Now returns a strictly increasing timestamp. Timers retain their scheduled
// deadline separately, so a delayed manual fire cannot be mistaken for the
// clock's later read time.
func (c *manualAttemptClock) Now() time.Time {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.now = c.now.Add(time.Millisecond)
	return c.now
}
func (c *manualAttemptClock) current() time.Time {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.now
}
func (c *manualAttemptClock) advanceTo(at time.Time) {
	c.mu.Lock()
	if c.now.Before(at) {
		c.now = at
	}
	c.mu.Unlock()
}
func (c *manualAttemptClock) NewTimer(d time.Duration) attemptTimer {
	timer := newManualAttemptTimer(d, c.current, c.advanceTo)
	c.mu.Lock()
	c.timers = append(c.timers, timer)
	beforeTimerReturn := c.beforeTimerReturn
	c.mu.Unlock()
	c.created <- struct{}{}
	if beforeTimerReturn != nil {
		beforeTimerReturn(timer)
	}
	return timer
}
func (c *manualAttemptClock) waitTimer(t *testing.T, index int) *manualAttemptTimer {
	t.Helper()
	for {
		c.mu.Lock()
		if len(c.timers) > index {
			timer := c.timers[index]
			c.mu.Unlock()
			return timer
		}
		c.mu.Unlock()
		select {
		case <-c.created:
		case <-time.After(2 * time.Second):
			t.Fatalf("timer %d was not created", index)
		}
	}
}
func (c *manualAttemptClock) count() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return len(c.timers)
}

type controlledRunCall struct {
	ctx  context.Context
	spec runtime.ExecutionSpec
	sink runtime.EventSink
}

type controlledTunnelCall struct {
	ctx  context.Context
	req  runtime.ProviderTunnelRequest
	sink runtime.ProviderTunnelSink
}

type controlledWatchdogAdapter struct {
	name          string
	runCalls      chan controlledRunCall
	tunnelCalls   chan controlledTunnelCall
	runReturn     chan error
	tunnelReturn  chan error
	maxConcurrent int
}

func newControlledWatchdogAdapter(name string) *controlledWatchdogAdapter {
	return &controlledWatchdogAdapter{
		name: name, runCalls: make(chan controlledRunCall, 1), tunnelCalls: make(chan controlledTunnelCall, 1),
		runReturn: make(chan error, 1), tunnelReturn: make(chan error, 1), maxConcurrent: 1,
	}
}
func (a *controlledWatchdogAdapter) Name() string { return a.name }
func (a *controlledWatchdogAdapter) Capabilities(context.Context) (runtime.Capabilities, error) {
	return runtime.Capabilities{AdapterName: a.name, Targets: []string{"target"}, MaxConcurrency: a.maxConcurrent}, nil
}
func (a *controlledWatchdogAdapter) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runtime.EventSink) error {
	a.runCalls <- controlledRunCall{ctx: ctx, spec: spec, sink: sink}
	return <-a.runReturn
}
func (a *controlledWatchdogAdapter) TunnelProvider(ctx context.Context, req runtime.ProviderTunnelRequest, sink runtime.ProviderTunnelSink) error {
	a.tunnelCalls <- controlledTunnelCall{ctx: ctx, req: req, sink: sink}
	return <-a.tunnelReturn
}

// probeCall records one invocation of the injected health probe so tests can
// assert the probe received an independent, still-live context after the
// stalled request was canceled.
type probeCall struct {
	ctx    context.Context
	target string
}

type probeReply struct {
	result runtime.ProviderProbeResult
	err    error
}

// probingWatchdogAdapter is a controlledWatchdogAdapter that also implements
// runtime.ProviderProber. ProbeProvider blocks on a channel so tests drive the
// independent bounded health probe deterministically and observe the context it
// received.
type probingWatchdogAdapter struct {
	*controlledWatchdogAdapter
	probeCalls  chan probeCall
	probeReturn chan probeReply
}

func newProbingWatchdogAdapter(name string) *probingWatchdogAdapter {
	return &probingWatchdogAdapter{
		controlledWatchdogAdapter: newControlledWatchdogAdapter(name),
		probeCalls:                make(chan probeCall, 1),
		probeReturn:               make(chan probeReply, 1),
	}
}

func (a *probingWatchdogAdapter) ProbeProvider(ctx context.Context, target string) (runtime.ProviderProbeResult, error) {
	a.probeCalls <- probeCall{ctx: ctx, target: target}
	reply := <-a.probeReturn
	return reply.result, reply.err
}

type watchdogRouter struct{ adapter runtime.ProviderTunnelAdapter }

func (r *watchdogRouter) Resolve(_ context.Context, req runtime.RunRequest) (runtime.ExecutionSpec, error) {
	return runtime.ExecutionSpec{
		RunID: req.RunID, Adapter: r.adapter.Name(), Target: req.Target, SessionID: req.SessionID,
		Background: req.Background, Input: req.Input, TimeoutSec: req.TimeoutSec, Metadata: req.Metadata,
		ResponseStallTimeoutMS: req.ResponseStallTimeoutMS,
	}, nil
}
func (r *watchdogRouter) ResolveAdapter(ctx context.Context, req runtime.RunRequest) (runtime.ExecutionSpec, runtime.Provider, error) {
	spec, err := r.Resolve(ctx, req)
	return spec, r.adapter, err
}
func (r *watchdogRouter) LookupAdapter(name string) (runtime.Provider, error) {
	if name != r.adapter.Name() {
		return nil, fmt.Errorf("adapter %q not found", name)
	}
	return r.adapter, nil
}
func (r *watchdogRouter) GetAdapter(name string) (runtime.Provider, bool) {
	if name == r.adapter.Name() {
		return r.adapter, true
	}
	return nil, false
}

func newWatchdogNode(t *testing.T, adapter runtime.ProviderTunnelAdapter, clock *manualAttemptClock) *Node {
	t.Helper()
	st, err := store.New(":memory:", zap.NewNop())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = st.Close() })
	n := New("node-watchdog", &watchdogRouter{adapter: adapter}, st, 0, io.Discard, zap.NewNop(), nil)
	n.watchdogClock = clock
	return n
}

type watchdogPipe struct {
	edge   *toki.TcpClient
	sess   *transport.Session
	events chan *iop.RunEvent
	frames chan *iop.ProviderTunnelFrame
}

func newWatchdogPipe(t *testing.T) *watchdogPipe {
	t.Helper()
	edgeConn, nodeConn := net.Pipe()
	edgeParsers := toki.ParserMap{
		toki.TypeNameOf(&iop.RunEvent{}): func(b []byte) (proto.Message, error) {
			m := &iop.RunEvent{}
			return m, proto.Unmarshal(b, m)
		},
		toki.TypeNameOf(&iop.ProviderTunnelFrame{}): func(b []byte) (proto.Message, error) {
			m := &iop.ProviderTunnelFrame{}
			return m, proto.Unmarshal(b, m)
		},
	}
	edge := toki.NewTcpClient(edgeConn, 0, 0, edgeParsers)
	nodeClient := toki.NewTcpClient(nodeConn, 0, 0, toki.ParserMap{})
	pipe := &watchdogPipe{
		edge: edge, sess: transport.ExportNewSession(nodeClient, zap.NewNop(), "node-watchdog", "watchdog"),
		events: make(chan *iop.RunEvent, 8), frames: make(chan *iop.ProviderTunnelFrame, 8),
	}
	toki.AddListenerTyped[*iop.RunEvent](&edge.Communicator, func(event *iop.RunEvent) {
		pipe.events <- proto.Clone(event).(*iop.RunEvent)
	})
	toki.AddListenerTyped[*iop.ProviderTunnelFrame](&edge.Communicator, func(frame *iop.ProviderTunnelFrame) {
		pipe.frames <- proto.Clone(frame).(*iop.ProviderTunnelFrame)
	})
	t.Cleanup(func() { _ = edge.Close(); _ = nodeClient.Close() })
	return pipe
}

func waitContextCanceled(t *testing.T, ctx context.Context) {
	t.Helper()
	select {
	case <-ctx.Done():
	case <-time.After(2 * time.Second):
		t.Fatal("provider context was not canceled")
	}
}

func waitRunEvent(t *testing.T, events <-chan *iop.RunEvent) *iop.RunEvent {
	t.Helper()
	select {
	case event := <-events:
		return event
	case <-time.After(2 * time.Second):
		t.Fatal("run event was not emitted")
		return nil
	}
}

func waitTunnelFrame(t *testing.T, frames <-chan *iop.ProviderTunnelFrame) *iop.ProviderTunnelFrame {
	t.Helper()
	select {
	case frame := <-frames:
		return frame
	case <-time.After(2 * time.Second):
		t.Fatal("tunnel frame was not emitted")
		return nil
	}
}

func requireTimerDurations(t *testing.T, timer *manualAttemptTimer, want ...time.Duration) {
	t.Helper()
	got, _ := timer.snapshot()
	if len(got) != len(want) {
		t.Fatalf("timer durations = %v, want %v", got, want)
	}
	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("timer durations = %v, want %v", got, want)
		}
	}
}

func activeAdapterAttempts(n *Node, adapter string) int {
	n.adapterGatesMu.Lock()
	gate := n.adapterGates[adapter]
	n.adapterGatesMu.Unlock()
	if gate == nil {
		return 0
	}
	return gate.activeCount()
}

func TestAttemptObserverProgressResetsAndFenceIsMonotonic(t *testing.T) {
	clock := newManualAttemptClock()
	observer := newAttemptObserver(clock, time.Second)
	timer := clock.waitTimer(t, 0)
	observer.observe(runtime.DispositionNone)
	requireTimerDurations(t, timer, time.Second)
	observer.observe(runtime.DispositionProgress)
	requireTimerDurations(t, timer, time.Second, time.Second)
	timer.fire()
	expiry, valid := observer.expiryForSignal(<-observer.expired())
	if !valid || !observer.claimFence(expiry) || observer.claimFence(expiry) {
		t.Fatal("fence claim was not monotonic")
	}
	observer.observe(runtime.DispositionProgress)
	requireTimerDurations(t, timer, time.Second, time.Second)
}

func TestAttemptObserverCurrentArmSignalSurvivesImmediateFire(t *testing.T) {
	clock := newManualAttemptClock()
	clock.beforeTimerReturn = func(timer *manualAttemptTimer) { timer.fire() }
	observer := newAttemptObserver(clock, time.Nanosecond)

	expiry, valid := observer.expiryForSignal(<-observer.expired())
	if !valid {
		t.Fatal("current timer signal was rejected because expiry bookkeeping followed the fire")
	}
	if !observer.claimFence(expiry) {
		t.Fatal("current timer signal did not claim the fence")
	}
}

type recordingProtoSender struct {
	mu       sync.Mutex
	messages []proto.Message
	entered  chan struct{}
	release  chan struct{}
	once     sync.Once
}

func (s *recordingProtoSender) Send(message proto.Message) error {
	if s.entered != nil {
		s.once.Do(func() {
			close(s.entered)
			<-s.release
		})
	}
	s.mu.Lock()
	s.messages = append(s.messages, proto.Clone(message))
	s.mu.Unlock()
	return nil
}
func (s *recordingProtoSender) snapshot() []proto.Message {
	s.mu.Lock()
	defer s.mu.Unlock()
	return append([]proto.Message(nil), s.messages...)
}

func TestTunnelSinkStallClaimSerializesAcceptedFrame(t *testing.T) {
	clock := newManualAttemptClock()
	sender := &recordingProtoSender{entered: make(chan struct{}), release: make(chan struct{})}
	sink := &tunnelSink{sess: sender, observer: newAttemptObserver(clock, time.Second)}
	bodyDone := make(chan error, 1)
	go func() {
		bodyDone <- sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{Kind: runtime.ProviderTunnelFrameKindBody, Body: []byte("accepted")})
	}()
	<-sender.entered
	if sink.mu.TryLock() {
		sink.mu.Unlock()
		t.Fatal("tunnel emission lock was released before accepted frame Send completed")
	}
	close(sender.release)
	if err := <-bodyDone; err != nil {
		t.Fatal(err)
	}
	timer := clock.waitTimer(t, 0)
	timer.fire()
	expiry, valid := sink.observer.expiryForSignal(<-sink.observer.expired())
	if !valid || !sink.claimStall(expiry) {
		t.Fatal("stall claim failed after accepted frame completed")
	}
	if err := sink.emitClaimedTerminal(context.Background(), stalledTunnelFrame(runtime.ProviderTunnelRequest{RunID: "run", TunnelID: "tunnel"}, stallObservation{fence: "confirmed", idle: time.Second})); err != nil {
		t.Fatal(err)
	}
	if err := sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{Kind: runtime.ProviderTunnelFrameKindUsage}); err != nil {
		t.Fatal(err)
	}
	messages := sender.snapshot()
	if len(messages) != 2 {
		t.Fatalf("sent frames = %d, want body then terminal", len(messages))
	}
	if messages[0].(*iop.ProviderTunnelFrame).GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY || messages[1].(*iop.ProviderTunnelFrame).GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR {
		t.Fatalf("frame order = %v, %v", messages[0], messages[1])
	}
}

// TestTunnelWatchdogFinishFrameWithoutEndStallsOnce reproduces a provider
// stream that has emitted a logical OpenAI finish frame but has not closed its
// HTTP response. The finish frame is body progress, not a transport terminal:
// the watchdog must expire once and fence every later usage/END frame.
func TestTunnelWatchdogFinishFrameWithoutEndStallsOnce(t *testing.T) {
	clock := newManualAttemptClock()
	adapter := newControlledWatchdogAdapter("tunnel-finish-without-end")
	n := newWatchdogNode(t, adapter, clock)
	pipe := newWatchdogPipe(t)
	done := make(chan error, 1)
	go func() {
		done <- n.OnProviderTunnelRequest(context.Background(), pipe.sess, &iop.ProviderTunnelRequest{
			RunId: "tunnel-finish-without-end", TunnelId: "tunnel", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000,
		})
	}()
	call := <-adapter.tunnelCalls

	if err := call.sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{
		RunID: "tunnel-finish-without-end", TunnelID: "tunnel", Kind: runtime.ProviderTunnelFrameKindResponseStart,
		StatusCode: 200, Headers: map[string]string{"Content-Type": "text/event-stream"},
	}); err != nil {
		t.Fatal(err)
	}
	logicalFinish := []byte("data: {\"choices\":[{\"delta\":{\"content\":\"finished body\"},\"finish_reason\":\"stop\"}]}\n\n")
	if err := call.sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{
		RunID: "tunnel-finish-without-end", TunnelID: "tunnel", Kind: runtime.ProviderTunnelFrameKindBody, Body: logicalFinish,
	}); err != nil {
		t.Fatal(err)
	}
	if frame := waitTunnelFrame(t, pipe.frames); frame.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START {
		t.Fatalf("response start = %+v", frame)
	}
	if frame := waitTunnelFrame(t, pipe.frames); frame.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY || string(frame.GetBody()) != string(logicalFinish) {
		t.Fatalf("logical finish body = %+v", frame)
	}

	stallTimer := clock.waitTimer(t, 0)
	requireTimerDurations(t, stallTimer, time.Second, time.Second, time.Second)
	stallTimer.fire()
	waitContextCanceled(t, call.ctx)
	grace := clock.waitTimer(t, 1)
	adapter.tunnelReturn <- nil
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("tunnel result = %v", err)
	}
	terminal := waitTunnelFrame(t, pipe.frames)
	if terminal.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR ||
		terminal.GetFailure().GetCode() != string(runtime.FailureCodeResponseStalled) ||
		terminal.GetMetadata()["failure_code"] != string(runtime.FailureCodeResponseStalled) {
		t.Fatalf("stall terminal = %+v", terminal)
	}
	requireTimerDurations(t, grace, defaultAttemptCloseGrace)

	// These are the exact late frames an adapter can race after cancellation.
	// Both must be accepted as no-ops by the fenced sink.
	if err := call.sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{
		RunID: "tunnel-finish-without-end", TunnelID: "tunnel", Kind: runtime.ProviderTunnelFrameKindUsage,
		Usage: &runtime.UsageStats{InputTokens: 1, OutputTokens: 1},
	}); err != nil {
		t.Fatal(err)
	}
	if err := call.sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{
		RunID: "tunnel-finish-without-end", TunnelID: "tunnel", Kind: runtime.ProviderTunnelFrameKindEnd, End: true,
	}); err != nil {
		t.Fatal(err)
	}
	select {
	case frame := <-pipe.frames:
		t.Fatalf("late frame escaped the terminal fence: %+v", frame)
	default:
	}
}

func TestRunWatchdogStaleExpiryYieldsToProgress(t *testing.T) {
	clock := newManualAttemptClock()
	adapter := newControlledWatchdogAdapter("run-stale-expiry")
	n := newWatchdogNode(t, adapter, clock)
	pipe := newWatchdogPipe(t)
	done := make(chan error, 1)
	go func() {
		done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{RunId: "run-stale-expiry", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
	}()
	call := <-adapter.runCalls
	sink := call.sink.(*terminalDeferringSink)
	claimStarted := make(chan struct{})
	releaseClaim := make(chan struct{})
	claimResult := make(chan bool, 1)
	var firstClaim sync.Once
	sink.beforeStallClaim = func() {
		firstClaim.Do(func() {
			close(claimStarted)
			<-releaseClaim
		})
	}
	sink.afterStallClaim = func(claimed bool) { claimResult <- claimed }

	stallTimer := clock.waitTimer(t, 0)
	stallTimer.fire()
	<-claimStarted // The old timer was consumed before provider progress arrives.
	if err := call.sink.Emit(context.Background(), runtime.RuntimeEvent{RunID: "run-stale-expiry", Type: runtime.EventTypeDelta, Delta: "progress"}); err != nil {
		t.Fatal(err)
	}
	if event := waitRunEvent(t, pipe.events); event.GetType() != string(runtime.EventTypeDelta) {
		t.Fatalf("progress event = %+v", event)
	}
	requireTimerDurations(t, stallTimer, time.Second, time.Second)
	close(releaseClaim)
	if claimed := <-claimResult; claimed {
		t.Fatal("stale normalized expiry fenced after progress reset the watchdog")
	}
	select {
	case event := <-pipe.events:
		t.Fatalf("stale normalized expiry emitted terminal: %+v", event)
	default:
	}

	stallTimer.fire()
	waitContextCanceled(t, call.ctx)
	grace := clock.waitTimer(t, 1)
	adapter.runReturn <- nil
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("run result = %v", err)
	}
	if terminal := waitRunEvent(t, pipe.events); terminal.GetType() != string(runtime.EventTypeError) || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
		t.Fatalf("terminal = %+v", terminal)
	}
	if claimed := <-claimResult; !claimed {
		t.Fatal("reset normalized expiry did not claim the watchdog fence")
	}
	requireTimerDurations(t, grace, defaultAttemptCloseGrace)
}

func TestTunnelWatchdogStaleExpiryYieldsToProgress(t *testing.T) {
	clock := newManualAttemptClock()
	adapter := newControlledWatchdogAdapter("tunnel-stale-expiry")
	n := newWatchdogNode(t, adapter, clock)
	pipe := newWatchdogPipe(t)
	done := make(chan error, 1)
	go func() {
		done <- n.OnProviderTunnelRequest(context.Background(), pipe.sess, &iop.ProviderTunnelRequest{RunId: "tunnel-stale-expiry", TunnelId: "tunnel", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
	}()
	call := <-adapter.tunnelCalls
	sink := call.sink.(*tunnelSink)
	claimStarted := make(chan struct{})
	releaseClaim := make(chan struct{})
	claimResult := make(chan bool, 1)
	var firstClaim sync.Once
	sink.beforeStallClaim = func() {
		firstClaim.Do(func() {
			close(claimStarted)
			<-releaseClaim
		})
	}
	sink.afterStallClaim = func(claimed bool) { claimResult <- claimed }

	stallTimer := clock.waitTimer(t, 0)
	stallTimer.fire()
	<-claimStarted // The old timer was consumed before the accepted frame progresses the attempt.
	if err := call.sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{RunID: "tunnel-stale-expiry", TunnelID: "tunnel", Kind: runtime.ProviderTunnelFrameKindBody, Body: []byte("progress")}); err != nil {
		t.Fatal(err)
	}
	if frame := waitTunnelFrame(t, pipe.frames); frame.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY {
		t.Fatalf("progress frame = %+v", frame)
	}
	requireTimerDurations(t, stallTimer, time.Second, time.Second)
	close(releaseClaim)
	if claimed := <-claimResult; claimed {
		t.Fatal("stale tunnel expiry fenced after progress reset the watchdog")
	}
	select {
	case frame := <-pipe.frames:
		t.Fatalf("stale tunnel expiry emitted terminal: %+v", frame)
	default:
	}

	stallTimer.fire()
	waitContextCanceled(t, call.ctx)
	grace := clock.waitTimer(t, 1)
	adapter.tunnelReturn <- nil
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("tunnel result = %v", err)
	}
	if terminal := waitTunnelFrame(t, pipe.frames); terminal.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
		t.Fatalf("terminal = %+v", terminal)
	}
	if claimed := <-claimResult; !claimed {
		t.Fatal("reset tunnel expiry did not claim the watchdog fence")
	}
	requireTimerDurations(t, grace, defaultAttemptCloseGrace)
}

func TestRunWatchdogOldArmFireDuringResetYieldsToProgress(t *testing.T) {
	clock := newManualAttemptClock()
	adapter := newControlledWatchdogAdapter("run-old-arm-during-reset")
	n := newWatchdogNode(t, adapter, clock)
	pipe := newWatchdogPipe(t)
	done := make(chan error, 1)
	go func() {
		done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{RunId: "run-old-arm-during-reset", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
	}()
	call := <-adapter.runCalls
	sink := call.sink.(*terminalDeferringSink)
	captureResults := make(chan bool, 2)
	sink.observer.afterExpiryCapture = func(valid bool) { captureResults <- valid }

	stallTimer := clock.waitTimer(t, 0)
	clock.advanceTo(clock.current().Add(time.Second))
	stallTimer.beforeReset = stallTimer.fireStaleArmDuringReset
	if err := call.sink.Emit(context.Background(), runtime.RuntimeEvent{RunID: "run-old-arm-during-reset", Type: runtime.EventTypeDelta, Delta: "progress"}); err != nil {
		t.Fatal(err)
	}
	if event := waitRunEvent(t, pipe.events); event.GetType() != string(runtime.EventTypeDelta) {
		t.Fatalf("progress event = %+v", event)
	}
	if valid := <-captureResults; valid {
		t.Fatal("old normalized arm was accepted while progress reset the watchdog")
	}
	if err := call.ctx.Err(); err != nil {
		t.Fatal("old normalized arm canceled the provider before the reset threshold")
	}
	requireTimerDurations(t, stallTimer, time.Second, time.Second)

	stallTimer.fire()
	waitContextCanceled(t, call.ctx)
	grace := clock.waitTimer(t, 1)
	adapter.runReturn <- nil
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("run result = %v", err)
	}
	if valid := <-captureResults; !valid {
		t.Fatal("reset normalized arm was not accepted after its full threshold")
	}
	if terminal := waitRunEvent(t, pipe.events); terminal.GetType() != string(runtime.EventTypeError) || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
		t.Fatalf("terminal = %+v", terminal)
	}
	requireTimerDurations(t, grace, defaultAttemptCloseGrace)
}

func TestTunnelWatchdogOldArmFireDuringResetYieldsToProgress(t *testing.T) {
	clock := newManualAttemptClock()
	adapter := newControlledWatchdogAdapter("tunnel-old-arm-during-reset")
	n := newWatchdogNode(t, adapter, clock)
	pipe := newWatchdogPipe(t)
	done := make(chan error, 1)
	go func() {
		done <- n.OnProviderTunnelRequest(context.Background(), pipe.sess, &iop.ProviderTunnelRequest{RunId: "tunnel-old-arm-during-reset", TunnelId: "tunnel", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
	}()
	call := <-adapter.tunnelCalls
	sink := call.sink.(*tunnelSink)
	captureResults := make(chan bool, 2)
	sink.observer.afterExpiryCapture = func(valid bool) { captureResults <- valid }

	stallTimer := clock.waitTimer(t, 0)
	clock.advanceTo(clock.current().Add(time.Second))
	stallTimer.beforeReset = stallTimer.fireStaleArmDuringReset
	if err := call.sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{RunID: "tunnel-old-arm-during-reset", TunnelID: "tunnel", Kind: runtime.ProviderTunnelFrameKindBody, Body: []byte("progress")}); err != nil {
		t.Fatal(err)
	}
	if frame := waitTunnelFrame(t, pipe.frames); frame.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY {
		t.Fatalf("progress frame = %+v", frame)
	}
	if valid := <-captureResults; valid {
		t.Fatal("old tunnel arm was accepted while progress reset the watchdog")
	}
	if err := call.ctx.Err(); err != nil {
		t.Fatal("old tunnel arm canceled the provider before the reset threshold")
	}
	requireTimerDurations(t, stallTimer, time.Second, time.Second)

	stallTimer.fire()
	waitContextCanceled(t, call.ctx)
	grace := clock.waitTimer(t, 1)
	adapter.tunnelReturn <- nil
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("tunnel result = %v", err)
	}
	if valid := <-captureResults; !valid {
		t.Fatal("reset tunnel arm was not accepted after its full threshold")
	}
	if terminal := waitTunnelFrame(t, pipe.frames); terminal.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
		t.Fatalf("terminal = %+v", terminal)
	}
	requireTimerDurations(t, grace, defaultAttemptCloseGrace)
}

func TestRunWatchdogStaleExpiryBeforeCaptureYieldsToProgress(t *testing.T) {
	clock := newManualAttemptClock()
	adapter := newControlledWatchdogAdapter("run-stale-before-capture")
	n := newWatchdogNode(t, adapter, clock)
	pipe := newWatchdogPipe(t)
	done := make(chan error, 1)
	go func() {
		done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{RunId: "run-stale-before-capture", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
	}()
	call := <-adapter.runCalls
	sink := call.sink.(*terminalDeferringSink)
	captureStarted := make(chan struct{})
	releaseCapture := make(chan struct{})
	captureResults := make(chan bool, 4)
	var firstCapture sync.Once
	sink.observer.beforeExpiryCapture = func() {
		firstCapture.Do(func() {
			close(captureStarted)
			<-releaseCapture
		})
	}
	sink.observer.afterExpiryCapture = func(valid bool) { captureResults <- valid }

	stallTimer := clock.waitTimer(t, 0)
	stallTimer.fire()
	<-captureStarted // The old timer signal was received before its validity is captured.
	if err := call.sink.Emit(context.Background(), runtime.RuntimeEvent{RunID: "run-stale-before-capture", Type: runtime.EventTypeDelta, Delta: "progress"}); err != nil {
		t.Fatal(err)
	}
	if event := waitRunEvent(t, pipe.events); event.GetType() != string(runtime.EventTypeDelta) {
		t.Fatalf("progress event = %+v", event)
	}
	requireTimerDurations(t, stallTimer, time.Second, time.Second)
	close(releaseCapture)
	if valid := <-captureResults; valid {
		t.Fatal("stale normalized expiry captured as valid after progress reset the watchdog")
	}
	if err := call.ctx.Err(); err != nil {
		t.Fatal("stale normalized expiry canceled the provider before its reset threshold")
	}
	select {
	case event := <-pipe.events:
		t.Fatalf("stale normalized expiry emitted terminal: %+v", event)
	default:
	}

	stallTimer.fire()
	waitContextCanceled(t, call.ctx)
	grace := clock.waitTimer(t, 1)
	adapter.runReturn <- nil
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("run result = %v", err)
	}
	if valid := <-captureResults; !valid {
		t.Fatal("reset normalized expiry was not captured as valid after its full threshold")
	}
	if terminal := waitRunEvent(t, pipe.events); terminal.GetType() != string(runtime.EventTypeError) || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
		t.Fatalf("terminal = %+v", terminal)
	}
	requireTimerDurations(t, grace, defaultAttemptCloseGrace)
}

func TestTunnelWatchdogStaleExpiryBeforeCaptureYieldsToProgress(t *testing.T) {
	clock := newManualAttemptClock()
	adapter := newControlledWatchdogAdapter("tunnel-stale-before-capture")
	n := newWatchdogNode(t, adapter, clock)
	pipe := newWatchdogPipe(t)
	done := make(chan error, 1)
	go func() {
		done <- n.OnProviderTunnelRequest(context.Background(), pipe.sess, &iop.ProviderTunnelRequest{RunId: "tunnel-stale-before-capture", TunnelId: "tunnel", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
	}()
	call := <-adapter.tunnelCalls
	sink := call.sink.(*tunnelSink)
	captureStarted := make(chan struct{})
	releaseCapture := make(chan struct{})
	captureResults := make(chan bool, 4)
	var firstCapture sync.Once
	sink.observer.beforeExpiryCapture = func() {
		firstCapture.Do(func() {
			close(captureStarted)
			<-releaseCapture
		})
	}
	sink.observer.afterExpiryCapture = func(valid bool) { captureResults <- valid }

	stallTimer := clock.waitTimer(t, 0)
	stallTimer.fire()
	<-captureStarted // The old timer signal was received before its validity is captured.
	if err := call.sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{RunID: "tunnel-stale-before-capture", TunnelID: "tunnel", Kind: runtime.ProviderTunnelFrameKindBody, Body: []byte("progress")}); err != nil {
		t.Fatal(err)
	}
	if frame := waitTunnelFrame(t, pipe.frames); frame.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY {
		t.Fatalf("progress frame = %+v", frame)
	}
	requireTimerDurations(t, stallTimer, time.Second, time.Second)
	close(releaseCapture)
	if valid := <-captureResults; valid {
		t.Fatal("stale tunnel expiry captured as valid after progress reset the watchdog")
	}
	if err := call.ctx.Err(); err != nil {
		t.Fatal("stale tunnel expiry canceled the provider before its reset threshold")
	}
	select {
	case frame := <-pipe.frames:
		t.Fatalf("stale tunnel expiry emitted terminal: %+v", frame)
	default:
	}

	stallTimer.fire()
	waitContextCanceled(t, call.ctx)
	grace := clock.waitTimer(t, 1)
	adapter.tunnelReturn <- nil
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("tunnel result = %v", err)
	}
	if valid := <-captureResults; !valid {
		t.Fatal("reset tunnel expiry was not captured as valid after its full threshold")
	}
	if terminal := waitTunnelFrame(t, pipe.frames); terminal.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
		t.Fatalf("terminal = %+v", terminal)
	}
	requireTimerDurations(t, grace, defaultAttemptCloseGrace)
}
