fix(prompt): cancel yours and follow the right turn
Track submission membership through completion and report timeout cleanup honestly.
11 unmodified lines1213141516171819202122232425262728293031323334353637383940414243444511 unmodified lines
## asynchronous persistence failures
The dispatcher logs failures to advance delivered schedule occurrences or save mail delivery errors. Undelivered mail and unadvanced schedules remain durable and retry under the existing dispatch policy. A schedule whose submission succeeded but whose advance failed can be delivered again.
## Prompt ownership and cancellation
`POST /sessions/:id/events` accepts an optional `submissionId` alongside`clientId`. The CLI generates a new submission ID for each prompt. The sessionactor owns queued submissions and every input contributing to an active turn,including inputs without an ID.
`POST /sessions/:id/cancel-submission` takes `{ "submissionId": "..." }`.The actor returns `outcome` atomically:
- `cancelled_queued`: removed that submission from the queue.- `interrupt_requested`: requested interruption of its exclusively owned turn.- `shared_running`: other inputs contribute to the turn, which continues.- `not_pending`: the submission is neither queued nor active.
Cancellation never falls back to session-wide interruption. Repeating therequest is safe. An interruption acknowledgment does not confirm that workstopped. The CLI uses a separate five-second cleanup deadline and reports anunconfirmed cancellation when cleanup fails.
Ordered live `turn_membership` events carry `turnId` and `submissionIds`.Membership precedes worker output and expands when queued input steers a turn.`turn_completed` carries the same `turnId` after its final output or failure.Retries and compaction within the turn preserve this identity. These eventsare live bookkeeping, not durable transcript rows. Prompt clients follow theirsubmission through completion instead of inferring ownership from `clientId`or session idle status. A stream reset before completion leaves the outcomeuncertain. Daemon health advertises `submission_cancellation`; CLI promptsrequire that capability before submitting.177 unmodified linesExpand all17817918018118218318218318418518618718718818919019119219319441 unmodified lines177 unmodified linesExpand all end), Check = self(), {Reader, ReaderMonitor} = spawn_monitor(fun() -> State = sys:get_state(Pid), %% session_state.State currently has 19 fields: history is field 6. state = element(1, State), 20 = tuple_size(State), %% session_state.State currently has 20 fields: history is field 6. state = element(1, State), 21 = tuple_size(State), Check ! {checked, element(7, State) =:= none, contains(State, <<"REPLAY_SENTINEL:">>)} end), {Unloaded, Reachable} = receive {checked, A, B} -> {A, B} end, {Unloaded, Reachable} = receive {checked, A, B} -> {A, B}; {'DOWN', ReaderMonitor, process, Reader, Reason} -> error({state_inspection_failed, Reason}) end, receive {'DOWN', ReaderMonitor, process, Reader, normal} -> ok end, erlang:garbage_collect(Pid), Retained = binary_bytes(Pid), {binary, AfterEntries} = process_info(Pid, binary),41 unmodified lines12345678791011121315 unmodified lines293031323334353637383925 unmodified lines6566676869707172737475766977787179807374817576828379808184858687888990919293848586878889909194959396979896979899100101102991001011021031041051061071081091101111121131141151161171181191201211091101111121221231241251161261271181191201211221231241251261271281291301281291321301311321331341351361371381391401411 unmodified line143144145146142147148149150151152145146147148153154155156150151152157158159160161158162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197160161162163164165166167168169170171172173174175176177178179180181198199package app
import ( "context" "crypto/rand" "encoding/hex" "errors" "fmt" "sync" "slices" "time"
"albedo/cli/internal/daemon")15 unmodified lines conn, err := s.Connect(ctx) if err != nil { return PromptResult{}, err } if capabilityErr := daemon.CheckCapability(ctx, conn, "submission_cancellation", "prompt submission cancellation"); capabilityErr != nil { return PromptResult{}, capabilityErr } var choice modelChoice if model != "" { profiles, profileErr := daemon.ProviderProfiles(ctx, conn) if profileErr != nil {25 unmodified lines }
client := daemon.NewChatClient(conn, sessionID) answer, err := awaitReply(ctx, client, prompt) if err == nil { return PromptResult{SessionID: sessionID, Answer: answer}, nil } if _, uncertain := errors.AsType[*daemon.UncertainOutcomeError](err); uncertain { return PromptResult{}, fmt.Errorf("%w; the session is %s", err, sessionID) } switch { case errors.Is(ctx.Err(), context.DeadlineExceeded): return PromptResult{}, fmt.Errorf("timed out after %s and stopped the turn; the session is %s", timeout, sessionID) return PromptResult{}, fmt.Errorf("timed out after %s; %v; the session is %s", timeout, err, sessionID) case ctx.Err() != nil: return PromptResult{}, fmt.Errorf("interrupted; the session is %s", sessionID) return PromptResult{}, fmt.Errorf("interrupted; %v; the session is %s", err, sessionID) } if err != nil { return PromptResult{}, err return PromptResult{}, err } return PromptResult{SessionID: sessionID, Answer: answer}, nil}
// awaitReply sends prompt and answers the last assistant text of the turn it// starts once the session is idle again, or the error that ended the turn.func awaitReply(ctx context.Context, client *daemon.ChatClient, prompt string) (string, error) {// awaitReply follows the logical turn containing this submission. The actor// publishes membership before any worker events, and completion after them.func awaitReply(ctx context.Context, client *daemon.ChatClient, prompt string) (answer string, failure error) { var identity [16]byte if _, err := rand.Read(identity[:]); err != nil { return "", err } submissionID := hex.EncodeToString(identity[:]) streamCtx, cancel := context.WithCancel(ctx) defer cancel() var ( mu sync.Mutex started bool // our prompt has reached the transcript owned bool // the latest observed live turn belongs to this invocation answer string failure error ) // Cleanup uses its own deadline even when the operation's context is canceled. submitted := false defer func() { if ctx.Err() == nil { if failure == nil || !submitted { return } mu.Lock() interrupt := owned mu.Unlock() if interrupt { interruptCtx, stop := context.WithTimeout(context.Background(), 5*time.Second) defer stop() _, _ = client.Interrupt(interruptCtx) cleanupCtx, stop := context.WithTimeout(context.Background(), 5*time.Second) defer stop() outcome, err := client.CancelSubmission(cleanupCtx, submissionID) if err != nil { failure = fmt.Errorf("%w; cancellation unconfirmed: %v", failure, err) return } switch outcome { case "cancelled_queued": failure = fmt.Errorf("%w; queued submission cancelled", failure) case "interrupt_requested": failure = fmt.Errorf("%w; interruption requested for the turn", failure) case "shared_running": failure = fmt.Errorf("%w; the shared turn is still running", failure) case "not_pending": failure = fmt.Errorf("%w; submission is no longer pending", failure) } }() ready := make(chan struct{}) events := make(chan daemon.StreamEvent, 64) streamDone := make(chan error, 1) go func() { streamDone <- client.Stream(streamCtx, 0, func(event daemon.StreamEvent) error { mu.Lock() defer mu.Unlock() switch { case event.Type == daemon.EventReset: if event.Type == daemon.EventReset { select { case <-ready: default: close(ready) // the snapshot is read; later events are live close(ready) } case event.Replayed: case event.Type == daemon.EventUser: owned = event.ClientID == client.ClientID() started = started || owned case !owned: case event.Type == daemon.EventMessage: answer, failure = event.Text, nil case event.Type == daemon.EventError: failure = errors.New(event.Text) case event.Type == daemon.EventInterrupted: failure = errors.New("the turn was interrupted") case event.Type == daemon.EventRetry: failure = nil } if event.Replayed { return nil return nil } select { case events <- event: return nil case <-streamCtx.Done(): return streamCtx.Err() } }) }() select { case <-ready:1 unmodified line return "", fmt.Errorf("could not follow the session: %w", err) case <-ctx.Done(): return "", ctx.Err() } if _, err := client.Send(ctx, prompt, nil); err != nil { // A failed request can already have reached the actor. Cleanup always uses // its identity, so it cannot interrupt a different client's turn. submitted = true if _, err := client.SendSubmission(ctx, prompt, submissionID); err != nil { return "", err }
ticker := time.NewTicker(200 * time.Millisecond) defer ticker.Stop() idleBefore := false turnID := "" // Discard the initial reset already acknowledged before submission. initialReset := true for { mu.Lock() turnSeen := started mu.Unlock() select { case <-ctx.Done(): return "", ctx.Err() case err := <-streamDone: return "", fmt.Errorf("lost the session stream before the turn finished: %v", err) case <-ticker.C: case event := <-events: if event.Type == daemon.EventReset && initialReset { initialReset = false continue } if event.Type == daemon.EventReset { return "", errors.New("session stream reset before the submission completed; outcome uncertain") } if event.Type == daemon.EventTurnMembership && slices.Contains(event.SubmissionIDs, submissionID) { turnID = event.TurnID } if turnID == "" { continue } switch event.Type { case daemon.EventMessage: answer, failure = event.Text, nil case daemon.EventError: failure = errors.New(event.Text) case daemon.EventInterrupted: failure = errors.New("the turn was interrupted") case daemon.EventRetry: failure = nil case daemon.EventTurnCompleted: if event.TurnID != turnID { continue } if failure != nil { return "", failure } if answer == "" { return "", errors.New("the turn ended without an assistant reply") } return answer, nil } } if !turnSeen { continue } // Idle after our prompt reached the transcript means its turn is over; // idle twice in a row gives the stream time to deliver its last events. status, err := client.GetStatus(ctx) if err != nil { return "", err } if !status.Idle || !idleBefore { idleBefore = status.Idle continue } mu.Lock() defer mu.Unlock() switch { case failure != nil: return "", failure case answer == "": return "", errors.New("the turn ended without an assistant reply") } return answer, nil }}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138package app
// These protocol tests control stream/request ordering that real provider// timing cannot guarantee, including a cleanup response after caller cancellation.import ( "context" "encoding/json" "fmt" "net" "net/http" "net/http/httptest" "strings" "testing" "time"
"albedo/cli/internal/daemon")
func promptTestConnection(server *httptest.Server) *daemon.Connection { return daemon.NewConnection(daemon.ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, "")}
func TestPromptRejectsAnOlderDaemonBeforeCreatingASession(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { if request.URL.Path != "/health" { t.Errorf("unsupported prompt issued %s", request.URL.Path) } _, _ = writer.Write([]byte(`{"capabilities":[]}`)) })) defer server.Close() service := Service{Connect: func(context.Context) (*daemon.Connection, error) { return promptTestConnection(server), nil }} _, err := service.RunPrompt(t.Context(), PromptOptions{Prompt: "hello"}) if err == nil || !strings.Contains(err.Error(), "needs an update") { t.Fatalf("old daemon result: %v", err) }}
func TestPromptFollowsCombinedSubmissionThroughRetryAndCompletion(t *testing.T) { submitted := make(chan string, 1) server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { switch request.URL.Path { case "/sessions/test/events": var payload struct { SubmissionID string `json:"submissionId"` } if err := json.NewDecoder(request.Body).Decode(&payload); err != nil { t.Error(err) return } if payload.SubmissionID == "" { t.Error("missing submission ID") } submitted <- payload.SubmissionID writer.WriteHeader(http.StatusAccepted) _, _ = writer.Write([]byte(`{"ok":true,"queued":true}`)) case "/sessions/test/stream": _, _ = fmt.Fprint(writer, "data: {\"cursor\":0,\"events\":[{\"type\":\"reset\"}]}\n\n") writer.(http.Flusher).Flush() id := <-submitted events := []map[string]any{ {"type": "turn_membership", "turnId": "combined", "submissionIds": []string{id, "other"}}, {"type": "user", "clientId": "another-client", "text": "other contribution"}, {"type": "message", "text": "before retry"}, {"type": "error", "text": "temporary failure"}, {"type": "retry"}, {"type": "compacted", "evicted": 1, "summary": "summary"}, {"type": "message", "text": "combined final answer"}, {"type": "turn_completed", "turnId": "combined"}, {"type": "turn_membership", "turnId": "later", "submissionIds": []string{"other"}}, {"type": "message", "text": "later answer"}, } page, _ := json.Marshal(map[string]any{"cursor": 1, "events": events}) _, _ = fmt.Fprintf(writer, "data: %s\n\n", page) writer.(http.Flusher).Flush() <-request.Context().Done() default: t.Errorf("unexpected %s", request.URL.Path) writer.WriteHeader(http.StatusNotFound) } })) defer server.Close() ctx, cancel := context.WithTimeout(t.Context(), time.Second) defer cancel() answer, err := awaitReply(ctx, daemon.NewChatClient(promptTestConnection(server), "test"), "hello") if err != nil || answer != "combined final answer" { t.Fatalf("combined reply: %q %v", answer, err) }}
func TestPromptCancellationReportsSharedWorkAndCleanupFailure(t *testing.T) { for _, fixture := range []struct { name, outcome, want string status int }{ {"shared", "shared_running", "shared turn is still running", 200}, {"cleanup failure", "", "cancellation unconfirmed", 503}, } { t.Run(fixture.name, func(t *testing.T) { submitted := make(chan struct{}) cancelled := make(chan struct{}) ctx, cancel := context.WithCancel(t.Context()) defer cancel() server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { switch request.URL.Path { case "/sessions/test/events": writer.WriteHeader(http.StatusAccepted) _, _ = writer.Write([]byte(`{"ok":true,"queued":true}`)) close(submitted) case "/sessions/test/stream": _, _ = fmt.Fprint(writer, "data: {\"cursor\":0,\"events\":[{\"type\":\"reset\"}]}\n\n") writer.(http.Flusher).Flush() <-request.Context().Done() case "/sessions/test/cancel-submission": if request.Context().Err() != nil { t.Error("cleanup reused cancelled context") } writer.WriteHeader(fixture.status) _, _ = fmt.Fprintf(writer, `{"outcome":%q}`, fixture.outcome) close(cancelled) default: t.Errorf("unexpected operation: %s", request.URL.Path) writer.WriteHeader(404) } })) defer server.Close() go func() { <-submitted; cancel() }() _, err := awaitReply(ctx, daemon.NewChatClient(promptTestConnection(server), "test"), "hello") if err == nil || !strings.Contains(err.Error(), fixture.want) || strings.Contains(err.Error(), "stopped") { t.Fatalf("cancellation report: %v", err) } select { case <-cancelled: default: t.Error("cleanup not attempted") } }) }}110 unmodified linesExpand all111112113114115116117118119120121122123124125126127128129130131132133134135136137429 unmodified linesExpand all110 unmodified linesExpand all } return c.submitPayload(ctx, payload)}
func (c *ChatClient) SendSubmission(ctx context.Context, content, submissionID string) (*SendResult, error) { return c.submitPayload(ctx, map[string]any{"content": content, "submissionId": submissionID})}
func (c *ChatClient) CancelSubmission(ctx context.Context, submissionID string) (string, error) { result, err := RequestOperation[struct { Outcome string `json:"outcome"` }](ctx, c.conn, Operation{Name: "cancel submission", Method: http.MethodPost, Path: sessionPath(c.agentID, "/cancel-submission"), Body: map[string]string{"submissionId": submissionID}}) if err != nil { return "", err } switch result.Outcome { case "cancelled_queued", "interrupt_requested", "shared_running", "not_pending": return result.Outcome, nil default: return "", fmt.Errorf("invalid cancellation outcome %q", result.Outcome) }}
func (c *ChatClient) Continue(ctx context.Context) (*SendResult, error) { return c.submitPayload(ctx, map[string]any{ "type": "continue", })429 unmodified linesExpand all4 unmodified lines5678910111213115 unmodified linesExpand all12913013113213313413513213613713813914014114217 unmodified lines1601611621631581591641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433444 unmodified linespackage e2e
import ( "context" "encoding/json" "net/http" "net/url" "strings" "testing"115 unmodified linesExpand all
// Cancellation while queued must leave the already running turn alone.func TestPromptTimeoutWhileQueuedDoesNotInterruptAnotherTurn(t *testing.T) { entered, release := make(chan struct{}), make(chan struct{}) releaseClosed := false defer func() { if !releaseClosed { defer close(release) close(release) } }() profile := providerRoute(t, func(request map[string]any) string { if lastUserText(request) == "other turn" { close(entered) <-release17 unmodified lines status, statusErr := client.GetStatus(t.Context()) if statusErr != nil || status.Idle { t.Fatalf("queued cancellation stopped the other turn: %+v %v", status, statusErr) } // Stop the session explicitly so this stalled fixture cannot leak into other tests. if _, err := client.Interrupt(t.Context()); err != nil { if !strings.Contains(stderr, "queued submission cancelled") || strings.Contains(stderr, "stopped") { t.Fatalf("incorrect cancellation report: %s", stderr) } close(release) releaseClosed = true waitIdle(t, id, profile, 1) if got := len(suite.provider.requests(profile)); got != 1 { t.Fatalf("cancelled queued prompt still ran: %d requests", got) } outcome, err := client.CancelSubmission(t.Context(), "missing-submission") if err != nil || outcome != "not_pending" { t.Fatalf("missing cancellation: %q %v", outcome, err) }}
// Both identified submissions contribute to the same next turn. Targeted// cancellation must leave that shared turn running, including anonymous input.func TestSubmissionCancellationLeavesSharedRunningTurn(t *testing.T) { firstEntered, firstRelease := make(chan struct{}), make(chan struct{}) sharedEntered, sharedRelease := make(chan struct{}), make(chan struct{}) firstReleased, sharedReleased := false, false defer func() { if !firstReleased { close(firstRelease) } if !sharedReleased { close(sharedRelease) } }() profile := providerRoute(t, func(request map[string]any) string { if lastUserText(request) == "blocking turn" { close(firstEntered) <-firstRelease } else { close(sharedEntered) <-sharedRelease } return echoReply(request) }) id := newSession(t, t.TempDir()) client := daemon.NewChatClient(conn(t), id) if _, err := client.Send(t.Context(), "blocking turn", nil); err != nil { t.Fatal(err) } select { case <-firstEntered: case <-time.After(10 * time.Second): t.Fatal("first turn never started") } if _, err := client.SendSubmission(t.Context(), "identified contribution", "shared-submission"); err != nil { t.Fatal(err) } if _, err := client.Send(t.Context(), "anonymous contribution", nil); err != nil { t.Fatal(err) } if _, err := client.SendSubmission(t.Context(), "cancel this queued contribution", "queued-submission"); err != nil { t.Fatal(err) } outcome, cancelErr := client.CancelSubmission(t.Context(), "queued-submission") if cancelErr != nil || outcome != "cancelled_queued" { t.Fatalf("queued cancellation = %q %v", outcome, cancelErr) } outcome, cancelErr = client.CancelSubmission(t.Context(), "queued-submission") if cancelErr != nil || outcome != "not_pending" { t.Fatalf("repeated queued cancellation = %q %v", outcome, cancelErr) } close(firstRelease) firstReleased = true select { case <-sharedEntered: case <-time.After(10 * time.Second): t.Fatal("shared turn never started") } for range 2 { outcome, err := client.CancelSubmission(t.Context(), "shared-submission") if err != nil || outcome != "shared_running" { t.Fatalf("shared cancellation = %q, %v", outcome, err) } } status, err := client.GetStatus(t.Context()) if err != nil || status.Idle { t.Fatalf("shared turn stopped: %+v %v", status, err) } close(sharedRelease) sharedReleased = true waitIdle(t, id, profile, 2) requests := suite.provider.requests(profile) encoded, marshalErr := json.Marshal(requests[1]) if marshalErr != nil || strings.Contains(string(encoded), "cancel this queued contribution") || !strings.Contains(string(encoded), "identified contribution") || !strings.Contains(string(encoded), "anonymous contribution") { t.Fatalf("queued cancellation damaged sibling inputs: %s %v", encoded, marshalErr) }}
// Cancellation races turn admission and completion. Once that target settles,// repeating cancellation cannot stop a later exclusively owned turn.func TestSubmissionCancellationRacesStartAndCompletion(t *testing.T) { firstEntered, firstRelease := make(chan struct{}), make(chan struct{}) laterEntered, laterRelease := make(chan struct{}), make(chan struct{}) firstReleased, laterReleased := false, false defer func() { if !firstReleased { close(firstRelease) } if !laterReleased { close(laterRelease) } }() profile := providerRoute(t, func(request map[string]any) string { switch lastUserText(request) { case "hold first turn": close(firstEntered) <-firstRelease case "hold later turn": close(laterEntered) <-laterRelease } return echoReply(request) }) id := newSession(t, t.TempDir()) client := daemon.NewChatClient(conn(t), id) if _, err := client.Send(t.Context(), "hold first turn", nil); err != nil { t.Fatal(err) } select { case <-firstEntered: case <-time.After(10 * time.Second): t.Fatal("first turn never started") } const submissionID = "racing-target" if _, err := client.SendSubmission(t.Context(), "short target turn", submissionID); err != nil { t.Fatal(err) } type cancellation struct { err error outcome string } result := make(chan cancellation, 1) start := make(chan struct{}) go func() { <-start outcome, err := client.CancelSubmission(t.Context(), submissionID) result <- cancellation{err: err, outcome: outcome} }() close(start) close(firstRelease) firstReleased = true var cancelled cancellation select { case cancelled = <-result: case <-time.After(10 * time.Second): t.Fatal("racing cancellation did not return") } if cancelled.err != nil { t.Fatal(cancelled.err) } switch cancelled.outcome { case "cancelled_queued", "interrupt_requested", "not_pending": default: t.Fatalf("unexpected racing cancellation outcome %q", cancelled.outcome) } waitIdle(t, id, profile, 1) if _, err := client.SendSubmission(t.Context(), "hold later turn", "later-owner"); err != nil { t.Fatal(err) } select { case <-laterEntered: case <-time.After(10 * time.Second): t.Fatal("later turn never started") } outcome, err := client.CancelSubmission(t.Context(), submissionID) if err != nil || outcome != "not_pending" { t.Fatalf("settled target cancellation = %q %v", outcome, err) } status, err := client.GetStatus(t.Context()) if err != nil || status.Idle { t.Fatalf("old cancellation stopped a later turn: %+v %v", status, err) } close(laterRelease) laterReleased = true waitIdle(t, id, profile, 2)}1723 unmodified linesExpand all1724172517261727172817281729173017311732173317341735173617371738173917401741174217431744174517468 unmodified lines17551756175717581749175917601761176217631764381 unmodified linesExpand all214621472148214921502151215221532154348 unmodified linesExpand all250325042505250624952496250725082509251025112512251325142515251625172518125 unmodified linesExpand all26442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675206 unmodified linesExpand all1723 unmodified linesExpand all }}
type Submission { User(text: String, client_id: String, image: Option(types.Image)) User( text: String, client_id: String, image: Option(types.Image), submission_id: Option(String), ) Continue(client_id: String)}
fn submission_decoder() -> decode.Decoder(Submission) { use submission_id <- decode.optional_field( "submissionId", None, decode.optional(decode.string), ) use client_id <- decode.optional_field("clientId", "", decode.string) use kind <- decode.optional_field("type", "user", decode.string) use content <- decode.optional_field( "content",8 unmodified lines case validate_submitted_image(submitted_image) { Error(message) -> decode.failure(Continue(client_id), message) Ok(image) -> case kind, content { "user", Some(text) -> decode.success(User(text, client_id, image)) "user", Some(text) -> decode.success(User(text, client_id, image, submission_id)) "continue", _ | "resume", _ -> decode.success(Continue(client_id)) "user", None -> decode.failure(Continue(client_id), "user content") _, _ -> decode.failure(Continue(client_id), "user, continue, or resume") }381 unmodified linesExpand all #( "capabilities", json.array( [ "submission_cancellation", "session_provider", "session_workspace", "session_extensions", "global_extensions",348 unmodified linesExpand all case actor.call(registry, 5000, Lookup(id, _)) { Error(message) -> error(404, message) Ok(worker) -> { let outcome = case submission { User(text, client_id, image) -> session.submit(worker, text, client_id, image) User(text, client_id, image, submission_id) -> session.submit_identified( worker, text, client_id, image, submission_id, ) Continue(client_id) -> session.submit_continue(worker, client_id) } case outcome {125 unmodified linesExpand all reply(202, json.object([#("submitted", json.bool(True))])) Error(e) -> error(409, e) } } Post, "cancel-submission" -> { case body(req, { use id <- decode.field("submissionId", decode.string) decode.success(id) }) { Ok(id) -> case string.trim(id) == "" { True -> error(400, "submissionId must not be empty") False -> reply( 200, json.object([ #( "outcome", json.string(session.cancel_submission(worker, id)), ), ]), ) } Error(message) -> error(400, message) } } Post, "interrupt" -> reply( 200, json.object([206 unmodified linesExpand all46 unmodified lines47484950515253545547 unmodified lines103104105106107108109110111120 unmodified linesExpand all23223323423523623723823924010 unmodified lines25125225325425225525625725825931 unmodified lines2912922932942952962972982992 unmodified lines302303304305302306307308309310311312313314315316317318319320321322323324325326327328329330331332333334311335336337338339125 unmodified linesExpand all46546646746846947047147247346 unmodified lines520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570503571572573574575576577578579580581582583584585586587588589590591592516517593594595596597598599600601255 unmodified linesExpand all857858859860784861862863864865866867868869870871872873251 unmodified linesExpand all11251126112711281044112911301131113211331134105011351136113711381139280 unmodified linesExpand all1420142114221423133913401341134213431344142414251426142714281429143014311432143314341435143614371438299 unmodified linesExpand all1738173917401741165317421743174417451746174717481749175017511752175371 unmodified lines1825182618271828182918301831183218331834183518361837183818391840320 unmodified linesExpand all46 unmodified lines restart_text, "daemon", turn.Note("daemon restart"), None, None,)
const continue_prompt = "<system-notice>continue your unfinished task, by resuming the most recent intent.47 unmodified linespub type Message { Resume Abort(String) Submit(Submission, Subject(Result(Bool, SubmissionError))) CancelSubmission(String, Subject(String)) ReadCommands( Subject(Result(#(List(command.Command), command.Context), String)), ) ReadCatalogContext(Subject(#(String, String, String, runtime.Runtime)))120 unmodified linesExpand all None, latest_usage, activity, [], [], 0, event_buffer.new(), [], None,10 unmodified lines // Background jobs wake this session through the kernel's jobs route; the // registered closure lands a completion notice as an ordinary submit, so // the wake reuses the whole turn pipeline and busy answers itself. wakes_register(info.id, fn(display, text) { wake(self, Submission(display, text, "job", turn.JobWake, None)) wake(self, Submission(display, text, "job", turn.JobWake, None, None)) }) commands_register(info.id, fn(op) { command_op(self, host, info.id, op) }) live_register(info.id, self) mailbox_register(info.id, fn(letter) {31 unmodified lines mail.text(letter), letter.id, turn.Mail(letter.id, letter.kind), None, None, ), )}
2 unmodified lines text: String, client_id: String, image: Option(types.Image),) -> Result(Bool, SubmissionError) { call_submit(session, Submission(text, text, client_id, turn.Chat, image)) call_submit( session, Submission(text, text, client_id, turn.Chat, image, None), )}
pub fn submit_identified( session: Session, text: String, client_id: String, image: Option(types.Image), submission_id: Option(String),) -> Result(Bool, SubmissionError) { call_submit( session, Submission(text, text, client_id, turn.Chat, image, submission_id), )}
pub fn cancel_submission(session: Session, submission_id: String) -> String { actor.call(session, 5000, CancelSubmission(submission_id, _))}
pub fn submit_continue( session: Session, client_id: String,) -> Result(Bool, SubmissionError) { call_submit( session, Submission("", continue_prompt, client_id, turn.Continue, None), Submission("", continue_prompt, client_id, turn.Continue, None, None), )}
/// This session's materialized command catalog and the context that runs them.125 unmodified linesExpand all Submit(..) | ReadCommands(..) | ReadCatalogContext(..) | Interrupt(..) | CancelSubmission(..) | Status(..) | Read(..) | ReadContext(..) | ChangeModel(..)46 unmodified lines reply, runtime.peek_commands(state.host, state.info.id, state.info.cwd), ) ReadSelection(reply) -> answer(state, reply, model_selection(state.info)) CancelSubmission(id, reply) -> { let remaining = list.filter(state.steering, fn(submission) { submission.submission_id != Some(id) }) case list.length(remaining) != list.length(state.steering) { True -> answer( session_state.State(..state, steering: remaining), reply, "cancelled_queued", ) False -> case turn.running(state.activity), state.active_submissions { Some(run), [submission] if submission.submission_id == Some(id) -> { interrupt_kernel(state) let _ = process.send_after(state.self, 2500, Abort(run.id)) answer( session_state.State( ..state, activity: turn.cancel(state.activity), ), reply, "interrupt_requested", ) } Some(_), submissions -> answer( state, reply, case list.any(submissions, fn(submission) { submission.submission_id == Some(id) }) { True -> "shared_running" False -> "not_pending" }, ) _, _ -> answer(state, reply, "not_pending") } } } Interrupt(reply) -> case turn.running(state.activity), waiting_turn(state) { // A turn still waiting for its kernel: drop it. Letters it held stay // undelivered, so the dispatcher offers them again later. None, True -> None, True -> { let id = mail.new_id() let state = session_state.State(..state, active_submissions: state.steering) |> session_submission.membership(id) |> session_state.emit(view.event("interrupted", [])) |> session_state.emit( view.event("turn_completed", [#("turnId", json.string(id))]), ) answer( session_state.State( ..state, steering: [], active_submissions: [], booting: option.map(state.booting, fn(boot) { #( boot.0, list.filter(boot.1, fn(work) { work != StartQueued && work != Resume }), ) }), ) |> session_state.emit(view.event("interrupted", [])), ), reply, True, ) } None, False -> answer(state, reply, False) Some(run), _ -> { interrupt_kernel(state) let _ = process.send_after(state.self, 2500, Abort(run.id))255 unmodified linesExpand all queued, timestamp, ) answer( session_state.State(..state, steering: []), session_state.State( ..state, steering: [], active_submissions: list.append( state.active_submissions, queued, ), ) |> session_submission.membership(id), reply, Ok(inputs), ) }251 unmodified linesExpand all command.ContextPage(section, page) -> context_page(session, section, page) command.Submit(display, text, client) -> submitted( session, Submission(display, text, client, turn.Chat, None), Submission(display, text, client, turn.Chat, None, None), "submitted", ) command.Note(origin, display, text) -> submitted( session, Submission(display, text, "", turn.Note(origin), None), Submission(display, text, "", turn.Note(origin), None, None), "queued", ) }}280 unmodified linesExpand all }}
fn failed_queued(state: State, error: String) -> State { session_state.emit( session_state.State(..state, steering: []), view.text( "error", "queued messages were not delivered; resend them: " <> error, ), let id = mail.new_id() session_state.State(..state, active_submissions: state.steering, steering: []) |> session_submission.membership(id) |> session_state.emit(view.text( "error", "queued messages were not delivered; resend them: " <> error, )) |> session_state.emit( view.event("turn_completed", [#("turnId", json.string(id))]), ) |> fn(state) { session_state.State(..state, active_submissions: []) }}
/// A child whose run ends without having answered its parent's latest task or/// message sends its last words, marked unreviewed, so nothing it found is299 unmodified linesExpand all let history = with_notice(history, state.notice) let state = session_history.remember(state, accepted, timestamp) |> session_submission.emit(submissions, timestamp) |> fn(state) { session_state.State(..state, notice: None, steering: []) } |> fn(state) { session_state.State( ..state, notice: None, steering: [], active_submissions: submissions, ) } Ok(#(state, kernel, client, history))}
fn start_now(71 unmodified lines let state = case run.work, run.cancelled { turn.Turn(_), False -> answer_parent(state, outcome) _, _ -> state } let state = case run.work { turn.Turn(_) -> session_state.emit( session_state.State(..state, active_submissions: []), view.event("turn_completed", [#("turnId", json.string(run.id))]), ) _ -> state } report_end(state, run) process.send(state.self, Collect) actor.continue(start_queued(follow_up(state)))}320 unmodified linesExpand all2 unmodified lines34567891011249 unmodified linesExpand all261262263264265266267268269270271272127 unmodified linesExpand all2 unmodified linesimport albedo/daemon/context_snapshotimport albedo/daemon/conversationimport albedo/daemon/requestsimport albedo/daemon/session_stateimport albedo/daemon/session_submissionimport albedo/daemon/transcriptimport albedo/daemon/turnimport albedo/daemon/usageimport albedo/harness/extension249 unmodified linesExpand all work: turn.Work, messages: Messages(message),) -> session_state.State(message) { let run_id = new_id() let state = case work { turn.Turn(_) -> session_submission.membership(state, run_id) _ -> state } let owner = state.self let stop = turn.latch() let client = stoppable(client, stop) // The worker's closures must capture these fields, never `state`: a spawn127 unmodified linesExpand all