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.
50 lines
1.8 KiB
Go
50 lines
1.8 KiB
Go
package output
|
|
|
|
import (
|
|
"encoding/json"
|
|
"io"
|
|
)
|
|
|
|
// InitEvent is the CLI-injected lifecycle event written at the head of a
|
|
// streaming command's NDJSON output. Carries enough context for an agent to
|
|
// thread follow-ups (session id, kb id, model, profile).
|
|
//
|
|
// Type field is always "init"; the JSON tag fixes the wire shape.
|
|
type InitEvent struct {
|
|
Type string `json:"type"`
|
|
SessionID string `json:"session_id"`
|
|
// MessageID anchors a resumed stream (`session continue-stream`) to the
|
|
// specific assistant message whose event buffer is being replayed. Empty
|
|
// for fresh streams (chat / session ask) where the message id is only
|
|
// known after the SDK emits its first agent_query frame.
|
|
MessageID string `json:"message_id,omitempty"`
|
|
AgentID string `json:"agent_id,omitempty"`
|
|
KBID string `json:"kb_id,omitempty"`
|
|
RequestID string `json:"request_id,omitempty"`
|
|
Model string `json:"model,omitempty"`
|
|
Profile string `json:"profile,omitempty"`
|
|
}
|
|
|
|
// WriteNDJSONLine emits one NDJSON event line to w.
|
|
func WriteNDJSONLine(w io.Writer, v any) error {
|
|
enc := json.NewEncoder(w)
|
|
enc.SetEscapeHTML(false)
|
|
return enc.Encode(v)
|
|
}
|
|
|
|
// EmitInit writes the lifecycle init event to start an NDJSON stream.
|
|
// MUST be called before any other writes to w within an NDJSON stream —
|
|
// agents key on lines[0].type == "init" to discover session/profile/kb
|
|
// context. Sets Type=init regardless of caller-supplied value.
|
|
func EmitInit(w io.Writer, ev InitEvent) error {
|
|
ev.Type = "init"
|
|
return WriteNDJSONLine(w, ev)
|
|
}
|
|
|
|
// EmitSDKEvent passes through the raw SDK event as one NDJSON line.
|
|
// The SDK is the source of truth for event vocabulary; the CLI does not
|
|
// rename or reshape events.
|
|
func EmitSDKEvent(w io.Writer, ev any) error {
|
|
return WriteNDJSONLine(w, ev)
|
|
}
|