nullkey
1bae6b6b6c
Adds `weknora session continue-stream <session-id> --message <id>` for
re-attaching to an in-progress or already-completed SSE event buffer.
Server semantics (replay-from-0 + tail):
- Every connection replays the full stored event log from index 0,
then tails any new events. NOT cursor-from-disconnect. Agents that
already consumed events on the original stream MUST dedupe by
message_id + event hash to avoid double-processing.
- Buffer TTL: redis mode 1h hardcoded; memory mode = process lifetime.
After expiry the CLI surfaces local.sse_stream_aborted.
Output is NDJSON: one CLI-injected init line carrying
{session_id, message_id, profile} at stream head, then raw SDK
StreamResponse events verbatim. The init line lets agents thread the
resume to the original message in their dedupe table before the first
SDK frame arrives — output.InitEvent gains an omitempty MessageID field
for this purpose; non-resume init events stay unchanged.
The command always emits NDJSON regardless of --format — there is no
human-text use case for raw event-log replay (operator scenarios are
incident response / debugging). --dry-run is excluded for the same
reason streaming commands always are: a buffered plan makes no sense
for an event stream.
75 lines
2.2 KiB
Go
75 lines
2.2 KiB
Go
package output_test
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"strings"
|
|
"testing"
|
|
|
|
"github.com/Tencent/WeKnora/cli/internal/output"
|
|
)
|
|
|
|
func TestEmitInit_ForcesTypeField(t *testing.T) {
|
|
var buf bytes.Buffer
|
|
// Pass an InitEvent with Type deliberately set to something wrong; EmitInit
|
|
// must overwrite it with "init".
|
|
ev := output.InitEvent{Type: "wrong", SessionID: "sess_x"}
|
|
if err := output.EmitInit(&buf, ev); err != nil {
|
|
t.Fatalf("EmitInit: %v", err)
|
|
}
|
|
var got struct {
|
|
Type string `json:"type"`
|
|
SessionID string `json:"session_id"`
|
|
}
|
|
if err := json.Unmarshal(buf.Bytes(), &got); err != nil {
|
|
t.Fatalf("unmarshal: %v\n%s", err, buf.String())
|
|
}
|
|
if got.Type != "init" {
|
|
t.Errorf("type: got %q, want init", got.Type)
|
|
}
|
|
if got.SessionID != "sess_x" {
|
|
t.Errorf("session_id: got %q, want sess_x", got.SessionID)
|
|
}
|
|
}
|
|
|
|
func TestEmitInit_OmitsEmptyOptionalFields(t *testing.T) {
|
|
var buf bytes.Buffer
|
|
// Only SessionID set; all optional fields should be absent from JSON.
|
|
ev := output.InitEvent{SessionID: "sess_only"}
|
|
if err := output.EmitInit(&buf, ev); err != nil {
|
|
t.Fatalf("EmitInit: %v", err)
|
|
}
|
|
line := buf.String()
|
|
for _, field := range []string{"kb_id", "agent_id", "model", "profile", "request_id", "message_id"} {
|
|
if strings.Contains(line, `"`+field+`"`) {
|
|
t.Errorf("optional field %q should be omitted when empty, got: %s", field, line)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestEmitSDKEvent_PassthroughVerbatim(t *testing.T) {
|
|
var buf bytes.Buffer
|
|
payload := map[string]any{"type": "answer", "content": "hi"}
|
|
if err := output.EmitSDKEvent(&buf, payload); err != nil {
|
|
t.Fatalf("EmitSDKEvent: %v", err)
|
|
}
|
|
var got map[string]any
|
|
if err := json.Unmarshal(buf.Bytes(), &got); err != nil {
|
|
t.Fatalf("unmarshal: %v\n%s", err, buf.String())
|
|
}
|
|
if got["type"] != "answer" {
|
|
t.Errorf("type: got %v, want answer", got["type"])
|
|
}
|
|
if got["content"] != "hi" {
|
|
t.Errorf("content: got %v, want hi", got["content"])
|
|
}
|
|
// Must be a single NDJSON line terminated by newline.
|
|
raw := buf.String()
|
|
if !strings.HasSuffix(raw, "\n") {
|
|
t.Errorf("NDJSON line must end with newline, got %q", raw)
|
|
}
|
|
if strings.Count(raw, "\n") != 1 {
|
|
t.Errorf("expected exactly one newline, got %d in %q", strings.Count(raw, "\n"), raw)
|
|
}
|
|
}
|