diff --git a/cli/src/ui/model-picker.test.tsx b/cli/src/ui/model-picker.test.tsx index 29ff363..2363b13 100644 --- a/cli/src/ui/model-picker.test.tsx +++ b/cli/src/ui/model-picker.test.tsx @@ -51,6 +51,10 @@ test("/model switches saved providers and models with manual fallback without lo if (rejectChange) { res.writeHead(409); res.end(JSON.stringify({ error: "session must be idle" })); return } changes.push(change); session.model = change.model if (change.provider) { session.provider = change.provider; session.protocol = change.provider === "other" ? "chat_completions" : "responses" } + const configuration = JSON.parse(await readFile(join(home, "config.json"), "utf8")) as { active: string; providers: Record } + configuration.active = session.provider + configuration.providers[session.provider]!.model = session.model + await writeFile(join(home, "config.json"), JSON.stringify(configuration)) res.end(JSON.stringify({ result: { model: session.model, provider: session.provider, protocol: session.protocol } })) } else if (req.url === "/sessions/session/commands") { res.end(JSON.stringify([])) @@ -148,5 +152,8 @@ test("/model switches saved providers and models with manual fallback without lo assert(!painted.includes("manual-model"), "old provider's model must not leak into the new catalog") stdin.write("\x1b") assert.equal(streams, 1, "switching providers must not discard the mounted conversation") - assert.equal(await readFile(join(home, "config.json"), "utf8"), config, "session selection must not change provider defaults") + const defaults = JSON.parse(await readFile(join(home, "config.json"), "utf8")) as { active: string; providers: Record } + assert.equal(defaults.active, "other") + assert.equal(defaults.providers.other?.model, "other-model") + assert.equal(defaults.providers.saved?.model, "manual-model") }) diff --git a/cli/src/ui/model-picker.tsx b/cli/src/ui/model-picker.tsx index 28a757e..2438a85 100644 --- a/cli/src/ui/model-picker.tsx +++ b/cli/src/ui/model-picker.tsx @@ -24,7 +24,7 @@ export function ModelPicker({ connection, provider, current, onSelect, onCancel }, [provider]) return albedo /model · {name || "session provider"} - changes this idle session only · provider defaults stay unchanged + changes this idle session and the default for new sessions {!saved ? <> {error || "loading providers…"} {}} onKey={(input, key) => { diff --git a/src/albedo/daemon/albedo_daemon.erl b/src/albedo/daemon/albedo_daemon.erl index 37f39f1..36d752b 100644 --- a/src/albedo/daemon/albedo_daemon.erl +++ b/src/albedo/daemon/albedo_daemon.erl @@ -1,11 +1,49 @@ -module(albedo_daemon). --export([env/1,ready/3,read_config/1,directory/1,shutdown/0,rss/1]). +-export([env/1,ready/3,read_config/1,write_default/3,directory/1,shutdown/0,rss/1]). env(Name) -> case os:getenv(binary_to_list(Name)) of false -> <<>>; Value -> unicode:characters_to_binary(Value) end. read_config(Home) -> case file:read_file(filename:join(Home,<<"config.json">>)) of {ok,Data} -> {ok,Data}; {error,_} -> {error,nil} end. +write_default(Home, Provider, Model) -> + File = filename:join(Home, <<"config.json">>), + Lock = filename:join(Home, <<"config.lock">>), + case file:open(Lock, [write, exclusive]) of + {error, _} -> {error, <<"provider configuration is being saved; retry">>}; + {ok, Owner} -> + try + case file:read_file(File) of + {ok, Bytes} -> + try + Config = json:decode(Bytes), + Named = case maps:find(<<"providers">>, Config) of + {ok, _} -> Config; + error -> #{<<"active">> => <<"default">>, <<"providers">> => #{<<"default">> => Config}} + end, + Providers = maps:get(<<"providers">>, Named), + Selected = maps:get(Provider, Providers), + Updated = Named#{<<"active">> => Provider, + <<"providers">> => Providers#{Provider => Selected#{<<"model">> => Model}}}, + Temp = <>, + case file:open(Temp, [write, exclusive, binary]) of + {ok, Handle} -> + try + ok = file:write(Handle, json:encode(Updated)), + ok = file:sync(Handle) + after file:close(Handle) end, + ok = file:change_mode(Temp, 8#600), + case file:rename(Temp, File) of + ok -> {ok, nil}; + {error, Reason} -> {error, atom_to_binary(Reason)} + end; + {error, Reason} -> {error, atom_to_binary(Reason)} + end + catch _:_ -> {error, <<"could not save provider default">>} end; + {error, _} -> {error, <<"provider configuration is missing">>} + end + after file:delete(<>), file:close(Owner), file:delete(Lock) end + end. directory(Path) -> filename:pathtype(Path) =:= absolute andalso filelib:is_dir(Path). ready(Home,Port,Token) -> File=filename:join(Home,<<"daemon.json">>), Temp= <>, diff --git a/src/albedo/daemon/configuration.gleam b/src/albedo/daemon/configuration.gleam index 9420db3..537913e 100644 --- a/src/albedo/daemon/configuration.gleam +++ b/src/albedo/daemon/configuration.gleam @@ -152,5 +152,20 @@ fn protocol_decoder() { }) } +pub fn select_default( + home: String, + provider: String, + model: String, +) -> Result(Nil, String) { + write_default(home, provider, model) +} + +@external(erlang, "albedo_daemon", "write_default") +fn write_default( + home: String, + provider: String, + model: String, +) -> Result(Nil, String) + @external(erlang, "albedo_daemon", "read_config") fn read_config(home: String) -> Result(BitArray, Nil) diff --git a/src/albedo/daemon/session.gleam b/src/albedo/daemon/session.gleam index fd3a88c..523e1e4 100644 --- a/src/albedo/daemon/session.gleam +++ b/src/albedo/daemon/session.gleam @@ -61,6 +61,16 @@ pub type SubmissionError { Rejected(String) } +fn valid_submission(display: String, text: String) -> Bool { + let maximum = case display == text { + True -> 1_048_576 + False -> 2_200_000 + } + string.trim(text) != "" + && string.byte_size(text) <= maximum + && string.byte_size(display) <= 1_048_576 +} + fn submission_error(error: SubmissionError) -> String { case error { WorkspaceMissing(path) -> "workspace not found: " <> path @@ -99,6 +109,7 @@ pub type Message { Commit(String, List(types.Input), String, Subject(Result(Int, String))) RecordContext(String, context_snapshot.Snapshot, Subject(Nil)) RecordUsage(String, usage.Metadata, Subject(Result(Nil, String))) + DrainSteering(String, Subject(Result(List(types.Input), String))) ReadContext(Subject(json.Json)) ReadContextPage(String, Int, Subject(Result(json.Json, String))) Finished(String, Result(Nil, String)) @@ -123,6 +134,7 @@ type State { history: Option(List(transcript.Entry)), latest_usage: Option(usage.Metadata), run: Option(Run), + steering: List(#(String, String, String, String, Option(types.Image))), sequence: Int, events: List(#(Int, String)), watchers: List(#(process.Pid, fn() -> Nil)), @@ -162,6 +174,7 @@ pub fn start( None, latest_usage, None, + [], 0, [], [], @@ -372,8 +385,30 @@ fn handle(state: State, message: Message) { Submit(display, text, client_id, source, image, reply) -> case state.run { Some(_) -> { - process.send(reply, Error(Rejected("session is busy"))) - actor.continue(state) + case + source == "chat" + && valid_submission(display, text) + && list.length(state.steering) < 32 + { + True -> { + process.send(reply, Ok(Nil)) + actor.continue( + State( + ..state, + steering: list.append(state.steering, [ + #(display, text, client_id, source, image), + ]), + ), + ) + } + False -> { + process.send( + reply, + Error(Rejected("session is busy or message queue is full")), + ) + actor.continue(state) + } + } } None -> { let maximum = case display == text { @@ -665,6 +700,11 @@ fn handle(state: State, message: Message) { "cannot switch provider: " <> error }), ) + use _ <- result.try(configuration.select_default( + state.home, + provider, + model, + )) use _ <- result.try(conversation.set_configuration( runtime.ledger(state.host), state.info.id, @@ -805,6 +845,68 @@ fn handle(state: State, message: Message) { actor.continue(state) } } + DrainSteering(id, reply) -> + case state.run { + Some(run) if run.id == id && !run.cancelled -> { + let inputs = + list.map(state.steering, fn(item) { + case item.4 { + Some(image) -> types.UserImage(item.1, image) + None -> types.User(item.1) + } + }) + case inputs { + [] -> { + process.send(reply, Ok([])) + actor.continue(state) + } + _ -> + case + conversation.commit_from( + runtime.ledger(state.host), + state.info.id, + inputs, + "model", + Some(state.info.provider), + ) + { + Error(error) -> { + process.send(reply, Error(error)) + actor.continue(state) + } + Ok(timestamp) -> { + let state = remember(state, inputs, timestamp) + let state = + list.fold(state.steering, state, fn(state, item) { + emit(state, case item.4 { + Some(image) -> + view.user_image( + item.0, + item.3, + Some(item.2), + Some(timestamp), + image, + ) + None -> + view.user( + item.0, + item.3, + Some(item.2), + Some(timestamp), + ) + }) + }) + process.send(reply, Ok(inputs)) + actor.continue(State(..state, steering: [])) + } + } + } + } + _ -> { + process.send(reply, Error("cancelled")) + actor.continue(state) + } + } RecordContext(id, snapshot, reply) -> { process.send(reply, Nil) case state.run { @@ -865,7 +967,7 @@ fn handle(state: State, message: Message) { _, _, Error(error) -> emit(state, view.text("error", error)) _, _, _ -> state } - actor.continue(state) + actor.continue(start_queued(state)) } _ -> actor.continue(state) } @@ -1222,6 +1324,80 @@ fn configured_client(state: State) -> Result(#(State, types.Client), String) { Ok(#(state, client)) } +fn failed_queued(state: State, error: String) -> State { + emit( + State(..state, steering: []), + view.text( + "error", + "queued messages were not delivered; resend them: " <> error, + ), + ) +} + +fn start_queued(state: State) -> State { + case state.steering { + [] -> state + queued -> + case prepare_submission(state) { + Error(error) -> failed_queued(state, submission_error(error)) + Ok(#(state, kernel, client)) -> { + let inputs = + list.map(queued, fn(item) { + case item.4 { + Some(image) -> types.UserImage(item.1, image) + None -> types.User(item.1) + } + }) + let accepted = list.append(recover_pending(state, kernel), inputs) + case projected_inputs(remember(state, accepted, 0)) { + Error(error) -> failed_queued(state, error) + Ok(history) -> + case + conversation.commit_from( + runtime.ledger(state.host), + state.info.id, + accepted, + "model", + Some(state.info.provider), + ) + { + Error(error) -> failed_queued(state, error) + Ok(timestamp) -> { + let state = remember(state, accepted, timestamp) + let state = + list.fold(queued, state, fn(state, item) { + emit(state, case item.4 { + Some(image) -> + view.user_image( + item.0, + item.3, + Some(item.2), + Some(timestamp), + image, + ) + None -> + view.user( + item.0, + item.3, + Some(item.2), + Some(timestamp), + ) + }) + }) + start_run( + State(..state, steering: [], notice: None), + kernel, + client, + history, + ) + } + } + } + } + } + } +} + fn start_run( state: State, kernel: runtime.Session, @@ -1261,6 +1437,7 @@ fn start_run( fn(metadata) { actor.call(owner, 10_000, RecordUsage(run_id, metadata, _)) }, + fn() { actor.call(owner, 10_000, DrainSteering(run_id, _)) }, ) let pid = process.spawn(fn() { diff --git a/src/albedo/harness/extensions/commands/extension.gleam b/src/albedo/harness/extensions/commands/extension.gleam index c53ab71..8c7955d 100644 --- a/src/albedo/harness/extensions/commands/extension.gleam +++ b/src/albedo/harness/extensions/commands/extension.gleam @@ -28,7 +28,7 @@ pub fn extension() -> extension.Extension { fn model() -> Command { Command( "/model", - "Show this session's provider and model, or switch the model. Switching is a user action and needs an idle session.", + "Show this session's provider and model, or switch it and make the selection the default for new sessions. Switching is a user action and needs an idle session.", [ Argument( "model", diff --git a/src/albedo/harness/extensions/python/extension.gleam b/src/albedo/harness/extensions/python/extension.gleam index 2161634..973a3c0 100644 --- a/src/albedo/harness/extensions/python/extension.gleam +++ b/src/albedo/harness/extensions/python/extension.gleam @@ -14,7 +14,7 @@ import gleam/result pub fn definition() -> types.Tool { types.Tool( "python", - "Execute Python in your persistent session. Top-level await works. Variables stay alive. Only printed output and the final expression return; truncated=true means the output passed the 64 KiB preview, and the rest is readable with output.read(cell_id, offset=65536). Failed cells are retained: await cells.read(id) for bounded source, await cells.info(id) for status without source, await cells.trace(id) for what the cell read, ran and changed. Repair exact unique text without resending the source with await cells.run(id, replacements=[(old,new)]), or pass check=True first to compile the rewrite and get the syntax error back without running anything. A cell that started requires allow_partial=True after checking side effects; never retry blindly.", + "Execute Python in your persistent session. Always provide both JSON arguments: code (string) and timeout_ms (integer, 1–3600000), even for a one-line call. Top-level await works. Variables stay alive. Only printed output and the final expression return; truncated=true means the output passed the 64 KiB preview, and the rest is readable with output.read(cell_id, offset=65536). Failed cells are retained: await cells.read(id) for bounded source, await cells.info(id) for status without source, await cells.trace(id) for what the cell read, ran and changed. Repair exact unique text without resending the source with await cells.run(id, replacements=[(old,new)]), or pass check=True first to compile the rewrite and get the syntax error back without running anything. A cell that started requires allow_partial=True after checking side effects; never retry blindly.", json.object([ #("type", json.string("object")), #("additionalProperties", json.bool(False)), diff --git a/src/albedo/harness/loop.gleam b/src/albedo/harness/loop.gleam index bc83015..97e6765 100644 --- a/src/albedo/harness/loop.gleam +++ b/src/albedo/harness/loop.gleam @@ -23,6 +23,7 @@ pub type Loop { commit: fn(List(types.Input), String) -> Result(Int, String), record_context: fn(types.Request) -> Nil, record_usage: fn(usage.Metadata) -> Result(Nil, String), + drain_steering: fn() -> Result(List(types.Input), String), ) } @@ -101,7 +102,22 @@ pub fn run( case turn.tool_calls { [] -> case turn.finish { - types.Complete -> Ok(Nil) + types.Complete -> { + use steering <- result.try(state.drain_steering()) + case steering { + [] -> Ok(Nil) + _ -> + run( + state, + id, + list.append( + list.reverse(steering), + list.append(list.reverse(replay), inputs), + ), + step + 1, + ) + } + } _ -> Error("model stopped: " <> string.inspect(turn.finish)) } calls -> { @@ -130,13 +146,17 @@ pub fn run( } }), ) + use steering <- result.try(state.drain_steering()) use _ <- result.try(state.commit([], "model")) run( state, id, list.append( - list.reverse(results), - list.append(list.reverse(replay), inputs), + list.reverse(steering), + list.append( + list.reverse(results), + list.append(list.reverse(replay), inputs), + ), ), step + 1, ) diff --git a/test/daemon/integration.py b/test/daemon/integration.py index 6114e26..ee5ff72 100644 --- a/test/daemon/integration.py +++ b/test/daemon/integration.py @@ -245,6 +245,27 @@ def run(protocol, endpoint): "gamma": provider("/gamma/v1", "gamma-1", "fixture-gamma", other_protocol), }) + # Busy chat messages steer at the next model boundary, after tool output. + queue_workspace=Path(directory)/"queue-workspace" + queue_workspace.mkdir() + queued=json.loads(cli("new", str(queue_workspace)))["session"] + request_start=len(Provider.requests) + with api(f"/sessions/{queued}/events", {"content":"first task"}): + pass + time.sleep(.15) + with api(f"/sessions/{queued}/events", {"content":"queued direction"}): + pass + with api(f"/sessions/{queued}/events", {"content":"another direction"}): + pass + ready(queued) + queued_requests=Provider.requests[request_start:] + assert len(queued_requests) == 2, queued_requests + assert "queued direction" not in json.dumps(queued_requests[0]["request"]), queued_requests + assert "queued direction" in json.dumps(queued_requests[1]["request"]), queued_requests + assert "another direction" in json.dumps(queued_requests[1]["request"]), queued_requests + assert json.dumps(queued_requests[1]["request"]).index("queued direction") < json.dumps(queued_requests[1]["request"]).index("another direction") + assert any(e.get("type") == "user" and e.get("text") == "queued direction" for e in snapshot(queued)), snapshot(queued) + # Workspace updates are idle-only and never interrupt active work. busy = json.loads(cli("new", str(busy_workspace)))["session"] with api(f"/sessions/{busy}/events", {"content":"hang while workspace update is attempted"}): @@ -319,10 +340,17 @@ def run(protocol, endpoint): with api(f"/sessions/{switched}/commands", {"name":"/model", "args":{"provider":"gamma", "model":"chosen-gamma"}}) as response: selection=json.load(response) assert selection == {"result": {"provider":"gamma", "model":"chosen-gamma", "protocol":other_protocol}}, selection + defaults=json.loads((home/"config.json").read_text()) + assert defaults["active"] == "gamma" and defaults["providers"]["gamma"]["model"] == "chosen-gamma", defaults + with api("/sessions", {"workspace": str(workspace)}) as response: + default_session=json.load(response) + assert default_session["provider"] == "gamma" and default_session["model"] == "chosen-gamma", default_session # Existing model-only callers keep the current provider and protocol. with api(f"/sessions/{switched}/commands", {"name":"/model", "args":{"model":"renamed-gamma"}}) as response: selection=json.load(response) assert selection == {"result": {"provider":"gamma", "model":"renamed-gamma", "protocol":other_protocol}}, selection + defaults=json.loads((home/"config.json").read_text()) + assert defaults["active"] == "gamma" and defaults["providers"]["gamma"]["model"] == "renamed-gamma", defaults try: api(f"/sessions/{switched}/commands", {"name":"/model", "args":{"provider":"unknown", "model":"wrong"}}).close() raise AssertionError("unknown provider switch succeeded") @@ -343,6 +371,8 @@ def run(protocol, endpoint): with api(f"/sessions/{switched}/commands", {"name":"/model", "args":{"provider":"alpha", "model":"returned-alpha"}}) as response: selection=json.load(response) assert selection == {"result": {"provider":"alpha", "model":"returned-alpha", "protocol":protocol}}, selection + defaults=json.loads((home/"config.json").read_text()) + assert defaults["active"] == "alpha" and defaults["providers"]["alpha"]["model"] == "returned-alpha", defaults request_start=len(Provider.requests) with api(f"/sessions/{switched}/events", {"content":"after second switch"}): pass @@ -421,6 +451,11 @@ def run(protocol, endpoint): assert next(item for item in listed if item["id"] == id)["title"] == "review the result" message_times = [(e["type"], e["text"], e.get("timestamp")) for e in snapshot(id) if e.get("type") in ("user", "message")] assert message_times and all(isinstance(stamp, int) for _, _, stamp in message_times) + configure("beta", { + "alpha": provider("/alpha/v1", "alpha-2", "returned-alpha"), + "beta": provider("/beta/v1", "beta-1", "fixture-beta"), + "gamma": provider("/gamma/v1", "gamma-1", "renamed-gamma", other_protocol), + }) # A single uninterrupted run can cross the former 100-model-turn ceiling. long_session=json.loads(cli("new",str(workspace)))["session"] request_start=len(Provider.requests)