package serve
import (
"context"
"strings"
"sync"
"time"
"codey/internal/acp"
"codey/internal/agent"
"codey/internal/approval"
)
// flushAfter and flushBytes are when buffered text goes out. One notification
// per token would be correct and unusable: T3 Code buffers inbound
// notifications in a queue 32 deep and drops the oldest when it overflows, so
// a fast model would lose the middle of its own sentence. Coalescing is what
// keeps us clear of that, and 50 ms is still below what reads as lag.
const (
flushAfter = 50 * time.Millisecond
flushBytes = 512
)
// The same idea for a tool that is still running, with a slower hand. Each
// update carries the whole output so far rather than a delta, so they cost
// more than a text chunk does and are worth less often: a progress view is
// read at a glance, not word by word.
//
// progressMaxBytes is the tail kept. The client bounds what it accepts at
// 8,000 characters and keeps the end; matching that here means the bound costs
// one truncation rather than two.
const (
progressAfter = 200 * time.Millisecond
progressBytes = 256
progressMaxBytes = 8000
)
// emitter turns one turn's [agent.Event] stream into session/update
// notifications.
//
// It is called from the agent's own goroutine, one event at a time, and holds
// a mutex anyway: the final flush comes from whichever goroutine ended the
// turn, and that is a different one.
type emitter struct {
send func(acp.Update)
now func() time.Time
afterTime time.Duration
afterSize int
// buf holds text that has not gone out yet, and bufKind says which kind
// of chunk it will become. Reasoning and the answer share the buffer but
// never share a chunk: appending one kind flushes the other first, because
// a client that got them merged would show the model thinking out loud in
// the middle of its answer.
mu sync.Mutex
buf strings.Builder
bufKind string
since time.Time
// anon names tool calls the model did not give an id to — the hosted
// tools, whose result blocks can arrive after a pause split them from
// their call. Dispatch is sequential, so the last one started is the one a
// result with no id belongs to.
anonN int
lastAnon string
// current is the call most recently announced, which is the one an
// approval prompt is about: the loop emits tool_start and then asks, one
// call at a time.
current string
// running accumulates the output of tool calls that have not finished,
// keyed by call id, with sentLen recording how much of each was last sent.
// Both are dropped when the call completes.
running map[string]*strings.Builder
sentLen map[string]int
lastProgress time.Time
}
// currentCall is the id of the tool call an approval prompt belongs to.
func (e *emitter) currentCall() string {
e.mu.Lock()
defer e.mu.Unlock()
return e.current
}
func newEmitter(send func(acp.Update)) *emitter {
return &emitter{
send: send,
now: time.Now,
afterTime: flushAfter,
afterSize: flushBytes,
running: make(map[string]*strings.Builder),
sentLen: make(map[string]int),
}
}
// handle is the [agent.Agent] observer.
func (e *emitter) handle(ev agent.Event) {
switch ev.Kind {
case agent.EventTextDelta:
e.appendText(acp.UpdateAgentMessageChunk, ev.Text)
case agent.EventText:
// The non-streaming path: the whole message at once.
e.Flush()
e.chunk(acp.UpdateAgentMessageChunk, ev.Text)
case agent.EventThinkingDelta:
e.appendText(acp.UpdateAgentThoughtChunk, ev.Text)
case agent.EventThinking:
e.Flush()
e.chunk(acp.UpdateAgentThoughtChunk, ev.Text)
case agent.EventToolStart:
// Anything buffered is said before the tool runs. Otherwise a sentence
// written just before a thirty-second command would sit in the buffer
// until the command finished, and the user would watch a tool run for
// no stated reason.
e.Flush()
e.toolCall(ev)
case agent.EventToolDelta:
e.toolProgress(ev)
case agent.EventToolResult:
e.Flush()
e.toolResult(ev)
case agent.EventNotice, agent.EventCompact:
// The loop talking about the run rather than about the task. It goes
// in the transcript because there is nowhere else for it to go, and
// silence here is how a compaction becomes a mystery.
if ev.Text != "" {
e.Flush()
e.chunk(acp.UpdateAgentMessageChunk, ev.Text)
}
}
}
func (e *emitter) appendText(kind, text string) {
if text == "" {
return
}
e.mu.Lock()
// A change of kind ends whatever was buffered, wherever it had got to.
var switched string
switchedKind := e.bufKind
if e.bufKind != kind && e.buf.Len() > 0 {
switched = e.takeLocked()
}
if e.buf.Len() == 0 {
e.since = e.now()
}
e.bufKind = kind
e.buf.WriteString(text)
var out string
if e.buf.Len() >= e.afterSize || e.now().Sub(e.since) >= e.afterTime {
out = e.takeLocked()
}
e.mu.Unlock()
if switched != "" {
e.chunk(switchedKind, switched)
}
if out != "" {
e.chunk(kind, out)
}
}
// Flush sends whatever text is buffered. It is safe to call when there is
// none, and must be called once more when the turn ends: the last delta of a
// message is usually short and recent, so nothing else would push it out.
func (e *emitter) Flush() {
e.mu.Lock()
kind := e.bufKind
out := e.takeLocked()
e.mu.Unlock()
if out != "" {
e.chunk(kind, out)
}
}
func (e *emitter) takeLocked() string {
out := e.buf.String()
e.buf.Reset()
return out
}
// chunkKindOr defends the one case takeLocked cannot: a flush with nothing
// buffered has no kind, and an untagged update is one a client drops.
func chunkKindOr(kind string) string {
if kind == "" {
return acp.UpdateAgentMessageChunk
}
return kind
}
func (e *emitter) chunk(kind, text string) {
e.send(acp.Update{SessionUpdate: chunkKindOr(kind), Content: acp.TextContent(text)})
}
func (e *emitter) toolCall(ev agent.Event) {
desc := approval.Describe(ev.Tool, ev.Input)
id := e.callID(ev, true)
e.mu.Lock()
e.current = id
e.mu.Unlock()
e.send(acp.Update{
SessionUpdate: acp.UpdateToolCall,
ToolCallID: id,
Title: title(desc),
Kind: toolKind(ev.Tool),
Status: acp.StatusInProgress,
RawInput: ev.Input,
})
}
// toolProgress reports output a tool is still producing.
//
// The protocol has no delta for this: a tool_call_update replaces the content
// of the call it names rather than appending to it, so each one carries
// everything so far. That is the shape clients expect — and defend against, by
// keeping only a bounded tail — so the bound is applied here too rather than
// making somebody else pay for our chattiness.
//
// Throttling is ours as well. A build writing a line at a time would otherwise
// be one notification per line, each carrying the whole log.
func (e *emitter) toolProgress(ev agent.Event) {
if ev.Text == "" {
return
}
id := e.callID(ev, false)
e.mu.Lock()
if e.running == nil {
e.running = make(map[string]*strings.Builder)
}
buf, ok := e.running[id]
if !ok {
buf = &strings.Builder{}
e.running[id] = buf
}
buf.WriteString(ev.Text)
grown := buf.Len() - e.sentLen[id]
due := grown >= progressBytes || e.now().Sub(e.lastProgress) >= progressAfter
var out string
if due {
out = tailOf(buf.String(), progressMaxBytes)
e.sentLen[id] = buf.Len()
e.lastProgress = e.now()
}
e.mu.Unlock()
if out == "" {
return
}
e.send(acp.Update{
SessionUpdate: acp.UpdateToolCallUpdate,
ToolCallID: id,
Status: acp.StatusInProgress,
Content: []acp.ToolCallContent{acp.ToolCallText(out)},
})
}
func (e *emitter) toolResult(ev agent.Event) {
status := acp.StatusCompleted
if ev.IsError {
status = acp.StatusFailed
}
id := e.callID(ev, false)
e.mu.Lock()
delete(e.running, id)
delete(e.sentLen, id)
e.mu.Unlock()
// The result replaces whatever progress said, which is what we want: the
// report a tool ends with is capped head and tail, where the live view was
// a tail alone.
e.send(acp.Update{
SessionUpdate: acp.UpdateToolCallUpdate,
ToolCallID: id,
Status: status,
Content: []acp.ToolCallContent{acp.ToolCallText(ev.Result)},
})
}
// tailOf keeps the end of a live-growing output, which is the part worth
// watching, and says that it did so.
func tailOf(s string, max int) string {
if len(s) <= max {
return s
}
return "[earlier output truncated]\n\n" + s[len(s)-max:]
}
func (e *emitter) callID(ev agent.Event, start bool) string {
if ev.ID != "" {
return ev.ID
}
e.mu.Lock()
defer e.mu.Unlock()
if start {
e.anonN++
e.lastAnon = "call-" + itoa(e.anonN)
}
if e.lastAnon == "" {
e.lastAnon = "call-0"
}
return e.lastAnon
}
// title is the one line a client shows beside the tool row. The summary
// already says what will happen; the detail is the argument that makes it
// specific, and a call without one is better off saying nothing than saying
// an empty pair of quotes.
func title(desc approval.Request) string {
detail := strings.TrimSpace(desc.Detail)
if detail == "" {
return desc.Summary
}
if i := strings.IndexByte(detail, '\n'); i >= 0 {
detail = detail[:i] + " …"
}
const max = 120
if len(detail) > max {
detail = detail[:max] + " …"
}
return desc.Summary + ": " + detail
}
// toolKind maps Codey's tools onto ACP's vocabulary, which a client uses to
// choose an icon and decide what to fold away. An unmapped tool is "other",
// which is a fine thing to be.
func toolKind(tool string) string {
switch tool {
case "read_file", "skill":
return acp.KindRead
case "glob", "grep":
return acp.KindSearch
case "edit_file", "write_file":
return acp.KindEdit
case "bash":
return acp.KindExecute
case "web_search", "web_fetch":
return acp.KindFetch
default:
return acp.KindOther
}
}
// itoa avoids pulling strconv in for one call site.
func itoa(n int) string {
if n == 0 {
return "0"
}
var b [20]byte
i := len(b)
for n > 0 {
i--
b[i] = byte('0' + n%10)
n /= 10
}
return string(b[i:])
}
// notifier is what an emitter sends through: the connection, bound to one
// session id.
func notifier(conn *acp.Conn, sessionID string) func(acp.Update) {
return func(u acp.Update) {
// A notification that cannot be written means the connection is gone,
// and the turn will find that out on its next move. There is nowhere
// useful to report it from here.
_ = conn.Notify(context.Background(), acp.MethodSessionUpdate,
acp.SessionNotification{SessionID: sessionID, Update: u})
}
}