package liveprotocol import ( "encoding/json" "errors" "reflect" "strings" "testing" "time" ) func TestFrameRoundTripCarriesReadyAndGoalControls(t *testing.T) { t.Parallel() frame := Frame{ ProtocolRevision: ProtocolRevision, StreamID: "stream-1", BindingID: "binding-1", Epoch: 7, Deliveries: []Delivery{ { Seq: 1, Kind: DeliveryKindStreamReady, StreamReady: &StreamReady{ ProtocolRevision: ProtocolRevision, StreamID: "stream-1", BindingID: "binding-1", }, }, { Seq: 2, Kind: DeliveryKindGoalChanged, GoalChanged: &GoalChanged{ WorkspaceID: "workspace-1", AgentSessionID: "session-1", Revision: 3, }, }, }, } encoded, err := EncodeFrame(frame) if err != nil { t.Fatal(err) } decoded, err := DecodeFrame(encoded) if err != nil { t.Fatal(err) } if decoded.Epoch != 7 || decoded.Deliveries[0].StreamReady.BindingID != "binding-1" || decoded.Deliveries[1].GoalChanged.Revision != 3 { t.Fatalf("decoded frame = %#v", decoded) } } func TestFrameRoundTripCarriesAttachmentRecoveryFence(t *testing.T) { t.Parallel() frame := Frame{ ProtocolRevision: ProtocolRevision, StreamID: "stream-1", BindingID: "binding-1", Epoch: 7, Deliveries: []Delivery{ { Seq: 1, Kind: DeliveryKindAttachmentChanged, AttachmentChanged: &AttachmentChanged{ BindingID: "binding-1", WorkspaceID: "workspace-1", AgentSessionID: "session-1", CanonicalTurnID: "canonical-turn-1", CanonicalTurnIDs: []string{"canonical-turn-1", "child-turn-1"}, CallerTurnID: "caller-turn-1", CurrentInteractionRootTurnID: "canonical-turn-1", AttachmentRevision: 3, }, }, { Seq: 2, Kind: DeliveryKindAttachmentCaughtUp, AttachmentCaughtUp: &AttachmentCaughtUp{ BindingID: "binding-1", WorkspaceID: "workspace-1", AgentSessionID: "session-1", CanonicalTurnID: "canonical-turn-1", CanonicalTurnIDs: []string{"canonical-turn-1", "child-turn-1"}, CallerTurnID: "caller-turn-1", CurrentInteractionRootTurnID: "canonical-turn-1", AttachmentRevision: 3, }, }, }, } encoded, err := EncodeFrame(frame) if err != nil { t.Fatal(err) } decoded, err := DecodeFrame(encoded) if err != nil { t.Fatal(err) } if changed := decoded.Deliveries[0].AttachmentChanged; changed == nil || changed.AttachmentRevision != 3 || changed.CallerTurnID != "caller-turn-1" || changed.CurrentInteractionRootTurnID != "canonical-turn-1" || len(changed.CanonicalTurnIDs) != 2 || changed.CanonicalTurnIDs[1] != "child-turn-1" { t.Fatalf("decoded attachment changed = %#v", changed) } if caughtUp := decoded.Deliveries[1].AttachmentCaughtUp; caughtUp == nil || caughtUp.AttachmentRevision != 3 || caughtUp.CanonicalTurnID != "canonical-turn-1" || caughtUp.CurrentInteractionRootTurnID != "canonical-turn-1" { t.Fatalf("decoded attachment caught up = %#v", caughtUp) } } func TestFrameRoundTripCarriesTurnlessGoalAuthorizationSet(t *testing.T) { t.Parallel() frame := Frame{ ProtocolRevision: ProtocolRevision, StreamID: "stream-1", BindingID: "binding-1", Epoch: 7, Deliveries: []Delivery{{ Seq: 1, Kind: DeliveryKindAttachmentChanged, AttachmentChanged: &AttachmentChanged{ BindingID: "binding-1", WorkspaceID: "workspace-1", AgentSessionID: "session-1", CanonicalTurnIDs: []string{"goal-turn-1", "goal-turn-2"}, CurrentInteractionRootTurnID: "goal-turn-2", AttachmentRevision: 3, }, }}, } encoded, err := EncodeFrame(frame) if err != nil { t.Fatal(err) } decoded, err := DecodeFrame(encoded) if err != nil { t.Fatal(err) } changed := decoded.Deliveries[0].AttachmentChanged if changed == nil || changed.CanonicalTurnID != "" || changed.CallerTurnID != "" || changed.CurrentInteractionRootTurnID != "goal-turn-2" || !reflect.DeepEqual(changed.CanonicalTurnIDs, []string{"goal-turn-1", "goal-turn-2"}) { t.Fatalf("decoded turnless Goal attachment = %#v", changed) } } func TestAttachmentRecoveryControlsRejectInvalidIdentityOrRevision(t *testing.T) { t.Parallel() for _, delivery := range []Delivery{ { Seq: 1, Kind: DeliveryKindAttachmentChanged, AttachmentChanged: &AttachmentChanged{ BindingID: "binding-1", WorkspaceID: "workspace-1", AgentSessionID: "session-1", }, }, { Seq: 1, Kind: DeliveryKindAttachmentCaughtUp, AttachmentCaughtUp: &AttachmentCaughtUp{ BindingID: "binding-1", WorkspaceID: "workspace-1", AgentSessionID: "session-1", }, }, { Seq: 1, Kind: DeliveryKindAttachmentChanged, AttachmentChanged: &AttachmentChanged{ BindingID: "binding-1", WorkspaceID: "workspace-1", AgentSessionID: "session-1", CanonicalTurnID: "canonical-turn-1", AttachmentRevision: 1, }, }, { Seq: 1, Kind: DeliveryKindAttachmentCaughtUp, AttachmentCaughtUp: &AttachmentCaughtUp{ BindingID: "binding-1", WorkspaceID: "workspace-1", AgentSessionID: "session-1", CallerTurnID: "caller-turn-1", AttachmentRevision: 1, }, }, { Seq: 1, Kind: DeliveryKindAttachmentChanged, AttachmentChanged: &AttachmentChanged{ BindingID: "binding-1", WorkspaceID: "workspace-1", AgentSessionID: "session-1", CanonicalTurnIDs: []string{"goal-turn-1", "goal-turn-1"}, AttachmentRevision: 1, }, }, { Seq: 1, Kind: DeliveryKindAttachmentChanged, AttachmentChanged: &AttachmentChanged{ BindingID: "binding-1", WorkspaceID: "workspace-1", AgentSessionID: "session-1", CanonicalTurnIDs: []string{"goal-turn-1", "goal-turn-2"}, AttachmentRevision: 1, }, }, { Seq: 1, Kind: DeliveryKindAttachmentCaughtUp, AttachmentCaughtUp: &AttachmentCaughtUp{ BindingID: "binding-1", WorkspaceID: "workspace-1", AgentSessionID: "session-1", CanonicalTurnIDs: []string{"goal-turn-1", "goal-turn-2"}, CurrentInteractionRootTurnID: "unproven-turn", AttachmentRevision: 1, }, }, { Seq: 1, Kind: DeliveryKindAttachmentChanged, AttachmentChanged: &AttachmentChanged{ BindingID: "binding-1", WorkspaceID: "workspace-1", AgentSessionID: "session-1", CanonicalTurnID: "canonical-turn-1", CanonicalTurnIDs: []string{"canonical-turn-1", "child-turn-1"}, CallerTurnID: "caller-turn-1", CurrentInteractionRootTurnID: "child-turn-1", AttachmentRevision: 1, }, }, } { if _, err := encodeDelivery(delivery); !errors.Is(err, ErrInvalidFrame) { t.Fatalf("encodeDelivery(%v) error = %v, want %v", delivery.Kind, err, ErrInvalidFrame) } } } func TestAttachmentRecoveryControlDecodeRequiresExplicitCurrentInteractionRoot(t *testing.T) { t.Parallel() for name, raw := range map[string][]byte{ "missing": []byte(`{"bindingId":"binding-1","workspaceId":"workspace-1","agentSessionId":"session-1","attachmentRevision":1}`), "null": []byte(`{"bindingId":"binding-1","workspaceId":"workspace-1","agentSessionId":"session-1","currentInteractionRootTurnId":null,"attachmentRevision":1}`), } { for _, target := range []any{&AttachmentChanged{}, &AttachmentCaughtUp{}} { if err := strictControlDecode(raw, target); !errors.Is(err, ErrInvalidFrame) { t.Fatalf("strictControlDecode(%T) %s root error = %v, want %v", target, name, err, ErrInvalidFrame) } } } // Empty is a meaningful value for a turnless attachment that has not yet // observed a Goal Turn; the field itself must still be present on the wire. explicitEmptyRoot := []byte(`{"bindingId":"binding-1","workspaceId":"workspace-1","agentSessionId":"session-1","currentInteractionRootTurnId":"","attachmentRevision":1}`) for _, target := range []any{&AttachmentChanged{}, &AttachmentCaughtUp{}} { if err := strictControlDecode(explicitEmptyRoot, target); err != nil { t.Fatalf("strictControlDecode(%T) explicit empty root error = %v", target, err) } } } func TestTypedRevisionRejectionDecodesAcrossRevisionMismatch(t *testing.T) { t.Parallel() frame := Frame{ ProtocolRevision: "sha256:older", StreamID: "stream-1", BindingID: "binding-1", Epoch: 1, Deliveries: []Delivery{{ Seq: 1, Kind: DeliveryKindRejected, Rejected: &Rejected{ Reason: RejectionProtocolRevisionMismatch, ExpectedRevision: ProtocolRevision, ReceivedRevision: "sha256:older", }, }}, } encoded, err := EncodeFrame(frame) if err != nil { t.Fatal(err) } decoded, err := DecodeFrame(encoded) if err != nil { t.Fatal(err) } if decoded.Deliveries[0].Rejected.Reason != RejectionProtocolRevisionMismatch { t.Fatalf("decoded rejection = %#v", decoded) } } func TestDecodeEventRejectsDuplicateAndUnknownFields(t *testing.T) { t.Parallel() duplicate := []byte(`{"workspaceId":"w","workspaceId":"w2","agentSessionId":"s","eventType":"message_delta","data":{}}`) if _, err := DecodeEvent(duplicate); !errors.Is(err, ErrInvalidLiveEvent) { t.Fatalf("duplicate error = %v", err) } event := mustMessageDelta(t, "message-1", &MessageContentOperation{ Operation: "set", Value: json.RawMessage(`"hello"`), }) raw, err := MarshalEvent(event) if err != nil { t.Fatal(err) } raw = []byte(strings.Replace(string(raw), `"eventType"`, `"unknown":true,"eventType"`, 1)) if _, err := DecodeEvent(raw); !errors.Is(err, ErrInvalidLiveEvent) { t.Fatalf("unknown-field error = %v", err) } } func TestPublisherCoalescesAdjacentAppendAndReplays(t *testing.T) { t.Parallel() now := time.Unix(100, 0) publisher, err := NewPublisher(PublisherConfig{ StreamID: "stream-1", BindingID: "binding-1", Epoch: 2, Now: func() time.Time { return now }, }) if err != nil { t.Fatal(err) } set := mustMessageDelta(t, "message-1", &MessageContentOperation{ Operation: "set", Value: json.RawMessage(`"H"`), }) if frames, err := publisher.Publish(PublishInput{Event: &set}); err != nil || len(frames) != 0 { t.Fatalf("set frames=%#v err=%v", frames, err) } for _, text := range []string{"el", "lo"} { appendEvent := mustMessageDelta(t, "message-1", &MessageContentOperation{ Operation: "append_text", Text: text, }) if frames, err := publisher.Publish(PublishInput{Event: &appendEvent}); err != nil || len(frames) != 0 { t.Fatalf("append frames=%#v err=%v", frames, err) } } frame, err := publisher.Flush() if err != nil { t.Fatal(err) } if frame == nil || len(frame.Deliveries) != 2 { t.Fatalf("frame = %#v, want set + coalesced append", frame) } _, appended, ok := messageAppendFromRaw(frame.Deliveries[1].Event) if !ok || appended.Content.Text != "ello" { t.Fatalf("coalesced append = %#v", appended) } resume, err := publisher.Resume(ResumeRequest{Epoch: 2, AfterSeq: 1}) if err != nil { t.Fatal(err) } if !resume.Hit || len(resume.Frames) != 1 || len(resume.Frames[0].Deliveries) != 1 || resume.Frames[0].Deliveries[0].Seq != 2 { t.Fatalf("resume = %#v", resume) } } func TestPublisherResumePartitionsReplayAtEncodedFrameBoundary(t *testing.T) { t.Parallel() const frameMaxBytes = 256 publisher, err := NewPublisher(PublisherConfig{ StreamID: "stream-1", BindingID: "binding-1", Epoch: 4, FrameMaxBytes: frameMaxBytes, }) if err != nil { t.Fatal(err) } for revision := int64(1); revision <= 12; revision++ { frames, err := publisher.Publish(PublishInput{ GoalChanged: &GoalChanged{ WorkspaceID: "workspace-1", AgentSessionID: "session-1", Revision: revision, }, Immediate: true, }) if err != nil || len(frames) != 1 { t.Fatalf("publish revision %d frames=%#v error=%v", revision, frames, err) } } resume, err := publisher.Resume(ResumeRequest{Epoch: 4}) if err != nil { t.Fatal(err) } if !resume.Hit || len(resume.Frames) < 2 { t.Fatalf("resume did not partition replay: %#v", resume) } var nextSeq uint64 = 1 for _, frame := range resume.Frames { encoded, err := EncodeFrame(frame) if err != nil { t.Fatal(err) } if len(encoded) > frameMaxBytes { t.Fatalf("encoded replay frame bytes=%d, max=%d", len(encoded), frameMaxBytes) } for _, delivery := range frame.Deliveries { if delivery.Seq != nextSeq { t.Fatalf("replay sequence=%d, want %d", delivery.Seq, nextSeq) } nextSeq++ } } if nextSeq != 13 { t.Fatalf("replayed through seq=%d, want 12", nextSeq-1) } } func TestSubscriberDetectsGapWithoutAcceptingLaterDelivery(t *testing.T) { t.Parallel() subscriber, err := NewSubscriber(SubscriberConfig{}) if err != nil { t.Fatal(err) } event := mustMessageDelta(t, "message-1", &MessageContentOperation{ Operation: "set", Value: json.RawMessage(`"hello"`), }) raw, _ := MarshalEvent(event) result := subscriber.Apply(Frame{ ProtocolRevision: ProtocolRevision, StreamID: "stream-1", BindingID: "binding-1", Epoch: 1, Deliveries: []Delivery{{ Seq: 2, Kind: DeliveryKindEvent, Event: raw, }}, }) if !result.ReconcileRequired || result.Reason != "sequence_gap" || len(result.Accepted) != 0 { t.Fatalf("apply = %#v", result) } } func TestSubscriberDropsDuplicateDelivery(t *testing.T) { t.Parallel() subscriber, err := NewSubscriber(SubscriberConfig{}) if err != nil { t.Fatal(err) } event := mustMessageDelta(t, "message-1", &MessageContentOperation{ Operation: "set", Value: json.RawMessage(`"hello"`), }) raw, _ := MarshalEvent(event) frame := Frame{ ProtocolRevision: ProtocolRevision, StreamID: "stream-1", BindingID: "binding-1", Epoch: 1, Deliveries: []Delivery{{Seq: 1, Kind: DeliveryKindEvent, Event: raw}}, } if result := subscriber.Apply(frame); len(result.Accepted) != 1 { t.Fatalf("first apply = %#v", result) } if result := subscriber.Apply(frame); result.DuplicateCount != 1 || len(result.Accepted) != 0 { t.Fatalf("duplicate apply = %#v", result) } } func TestPublisherReplacesOversizeEventWithScopedDiscontinuity(t *testing.T) { t.Parallel() publisher, err := NewPublisher(PublisherConfig{ StreamID: "stream-1", BindingID: "binding-1", Epoch: 1, DeliveryMaxBytes: 256, }) if err != nil { t.Fatal(err) } event := mustMessageDelta(t, "message-large", &MessageContentOperation{ Operation: "set", Value: json.RawMessage(`"` + strings.Repeat("x", 1024) + `"`), }) frames, err := publisher.Publish(PublishInput{Event: &event}) if err != nil { t.Fatal(err) } if len(frames) != 1 || len(frames[0].Deliveries) != 1 { t.Fatalf("frames = %#v", frames) } delivery := frames[0].Deliveries[0] if delivery.Seq != 1 || delivery.Kind != DeliveryKindDiscontinuity || delivery.Discontinuity.Reason != "delivery_too_large" || len(delivery.Discontinuity.ReconcileKeys) != 1 || delivery.Discontinuity.ReconcileKeys[0].MessageID != "message-large" { t.Fatalf("oversize delivery = %#v", delivery) } } func TestPublisherTerminalFenceFlushesAndRejectsLateTurnContent(t *testing.T) { t.Parallel() publisher, err := NewPublisher(PublisherConfig{ StreamID: "stream-1", BindingID: "binding-1", Epoch: 1, }) if err != nil { t.Fatal(err) } message := mustMessageDelta(t, "message-1", &MessageContentOperation{ Operation: "set", Value: json.RawMessage(`"hello"`), }) if frames, err := publisher.Publish(PublishInput{Event: &message}); err != nil || len(frames) != 0 { t.Fatalf("message frames=%#v err=%v", frames, err) } terminal := mustTerminalTurnUpdate(t) frames, err := publisher.Publish(PublishInput{Event: &terminal}) if err != nil { t.Fatal(err) } if len(frames) != 1 || len(frames[0].Deliveries) != 2 || frames[0].Deliveries[1].Seq != 2 { t.Fatalf("terminal frame = %#v", frames) } late := mustMessageDelta(t, "message-1", &MessageContentOperation{ Operation: "append_text", Text: " too late", }) frames, err = publisher.Publish(PublishInput{Event: &late}) if err != nil { t.Fatal(err) } if len(frames) != 1 || frames[0].Deliveries[0].Seq != 3 || frames[0].Deliveries[0].Kind != DeliveryKindDiscontinuity || frames[0].Deliveries[0].Discontinuity.Reason != "late_after_terminal" { t.Fatalf("late frame = %#v", frames) } } func TestRecipientProjectorPreservesNestedBusinessIdentity(t *testing.T) { t.Parallel() value := json.RawMessage(`{"workspaceId":"owner-workspace","agentSessionId":"owner-session","turnId":"owner-turn","path":"/same/sandbox/file"}`) event := mustMessageDelta(t, "message-1", &MessageContentOperation{ Operation: "set", Value: value, }) projector, err := NewRecipientProjector(ProjectionContext{ OwnerWorkspaceID: "owner-workspace", OwnerAgentSessionID: "owner-session", CanonicalTurnID: "owner-turn", RecipientWorkspaceID: "caller-workspace", RecipientAgentSessionID: "caller-session", CallerTurnID: "caller-turn", }) if err != nil { t.Fatal(err) } projected, err := projector.Project(event) if err != nil { t.Fatal(err) } var data MessageDeltaData if err := json.Unmarshal(projected.Data, &data); err != nil { t.Fatal(err) } if data.WorkspaceID != "caller-workspace" || data.AgentSessionID != "caller-session" || data.TurnID != "caller-turn" { t.Fatalf("projected identity = %#v", data) } var business map[string]any if err := json.Unmarshal(data.Content.Value, &business); err != nil { t.Fatal(err) } if business["workspaceId"] != "owner-workspace" || business["agentSessionId"] != "owner-session" || business["turnId"] != "owner-turn" || business["path"] != "/same/sandbox/file" { t.Fatalf("business payload was rewritten: %#v", business) } } func mustMessageDelta(t *testing.T, messageID string, operation *MessageContentOperation) Event { t.Helper() event, err := NewMessageDeltaEvent(MessageDeltaData{ WorkspaceID: "owner-workspace", AgentSessionID: "owner-session", MessageID: messageID, TurnID: "owner-turn", Role: "assistant", Kind: "text", OccurredAtUnixMS: 10, Content: operation, }) if err != nil { t.Fatal(err) } return event } func TestMessageToolOutputOperationValidation(t *testing.T) { t.Parallel() base := MessageDeltaData{ WorkspaceID: "workspace-1", AgentSessionID: "session-1", MessageID: "tool-1", TurnID: "turn-1", Role: "assistant", Kind: "tool_call", OccurredAtUnixMS: 10, } offset := int64(3) valid := base valid.ToolOutput = &MessageToolOutputOperation{ Operation: "append_text", Text: "好", OffsetBytes: &offset, } if _, err := NewMessageDeltaEvent(valid); err != nil { t.Fatalf("valid tool output rejected: %v", err) } for name, mutate := range map[string]func(*MessageDeltaData){ "missing offset": func(data *MessageDeltaData) { data.ToolOutput = &MessageToolOutputOperation{Operation: "append_text", Text: "x"} }, "wrong kind": func(data *MessageDeltaData) { data.Kind = "text" data.ToolOutput = &MessageToolOutputOperation{Operation: "set", Text: "x"} }, "set with offset": func(data *MessageDeltaData) { data.ToolOutput = &MessageToolOutputOperation{ Operation: "set", Text: "x", OffsetBytes: &offset, } }, } { t.Run(name, func(t *testing.T) { data := base mutate(&data) if _, err := NewMessageDeltaEvent(data); err == nil { t.Fatalf("invalid tool output accepted: %#v", data.ToolOutput) } }) } } func mustTerminalTurnUpdate(t *testing.T) Event { t.Helper() outcome := "completed" settledAt := int64(20) data := TurnUpdateData{ WorkspaceID: "owner-workspace", AgentSessionID: "owner-session", EventType: EventTypeTurnUpdate, OccurredAtUnixMS: 20, ActiveTurnID: nil, Turn: EventTurn{ TurnID: "owner-turn", AgentSessionID: "owner-session", Phase: "settled", Origin: "user_prompt", Outcome: &outcome, FileChanges: json.RawMessage(`null`), StartedAtUnixMS: 10, SettledAtUnixMS: &settledAt, UpdatedAtUnixMS: 20, }, } raw, err := json.Marshal(data) if err != nil { t.Fatal(err) } event := Event{ WorkspaceID: "owner-workspace", AgentSessionID: "owner-session", EventType: EventTypeTurnUpdate, Data: raw, } if _, err := MarshalEvent(event); err != nil { t.Fatal(err) } return event }