diff --git a/src/backend/backend.h b/src/backend/backend.h index 63b1aa1d..1e7924b3 100644 --- a/src/backend/backend.h +++ b/src/backend/backend.h @@ -112,6 +112,19 @@ struct _AiBackend GPtrArray *(*build_argv) (const AiBackend *self, const AiRunSpec *spec); + /* + * One turn, encoded for a CLI that reads its prompts from stdin. + * + * NULL means this backend takes its prompt in argv and exits when the turn + * is over, which is one process per turn. A backend that provides this is + * asked once and then kept: the process lives across turns, so whatever the + * agent left running -- a watch, a build, a background shell -- is still + * there for the next one. The caller owns the returned line and appends the + * newline itself. + */ + char *(*encode_turn) (const AiBackend *self, + const AiRunSpec *spec); + /* @root is one parsed line of output. */ void (*parse_object) (AiParser *parser, JsonObject *root, diff --git a/src/backend/claude-backend.c b/src/backend/claude-backend.c index 325d47b7..8330e9c4 100644 --- a/src/backend/claude-backend.c +++ b/src/backend/claude-backend.c @@ -3,7 +3,7 @@ /* * Claude Code, driven non-interactively. * - * claude -p --output-format stream-json --verbose + * claude -p --input-format stream-json --output-format stream-json --verbose * --include-partial-messages [--model M] [--append-system-prompt S] * [--resume SESSION] * @@ -11,6 +11,11 @@ * session and carries the id needed to resume it, "stream_event" lines carry * token deltas, and a final "result" line repeats the whole reply. * + * The prompt is not in argv: it arrives on stdin, one JSON line per turn, and + * the process stays up between them. That is what lets a background shell the + * agent started still be there on the next message -- with a prompt in argv the + * process ends when the turn does, and takes everything it started with it. + * * No permission flags are passed: in print mode the CLI already refuses tool * use by default, which is what we want for a chat window. */ @@ -46,8 +51,12 @@ claude_build_argv (const AiBackend *self, g_ptr_array_add (argv, g_strdup (spec->resume_session_id)); } + /* -p with nothing after it: the prompt comes over stdin instead, which is + * what keeps one process across the turns of a chat. */ g_ptr_array_add (argv, g_strdup ("-p")); - g_ptr_array_add (argv, g_strdup (spec->prompt)); + + g_ptr_array_add (argv, g_strdup ("--input-format")); + g_ptr_array_add (argv, g_strdup ("stream-json")); g_ptr_array_add (argv, g_strdup ("--output-format")); g_ptr_array_add (argv, g_strdup ("stream-json")); @@ -79,6 +88,46 @@ claude_build_argv (const AiBackend *self, return argv; } +/* + * One turn, as the CLI's streaming input expects it. + * + * The same envelope a user message has on the way out, which is what makes it + * readable: type "user", a message with a role and content parts. + */ +static char * +claude_encode_turn (const AiBackend *self, + const AiRunSpec *spec) +{ + g_autoptr (JsonBuilder) builder = json_builder_new (); + g_autoptr (JsonGenerator) generator = json_generator_new (); + g_autoptr (JsonNode) root = NULL; + + json_builder_begin_object (builder); + json_builder_set_member_name (builder, "type"); + json_builder_add_string_value (builder, "user"); + json_builder_set_member_name (builder, "message"); + json_builder_begin_object (builder); + json_builder_set_member_name (builder, "role"); + json_builder_add_string_value (builder, "user"); + json_builder_set_member_name (builder, "content"); + json_builder_begin_array (builder); + json_builder_begin_object (builder); + json_builder_set_member_name (builder, "type"); + json_builder_add_string_value (builder, "text"); + json_builder_set_member_name (builder, "text"); + json_builder_add_string_value ( + builder, spec->prompt != NULL ? spec->prompt : ""); + json_builder_end_object (builder); + json_builder_end_array (builder); + json_builder_end_object (builder); + json_builder_end_object (builder); + + root = json_builder_get_root (builder); + json_generator_set_root (generator, root); + + return json_generator_to_data (generator, NULL); +} + static void emit (AiEventFunc callback, gpointer user_data, @@ -520,5 +569,6 @@ const AiBackend xd_claude_backend = { .models = claude_models, .n_models = G_N_ELEMENTS (claude_models), .build_argv = claude_build_argv, + .encode_turn = claude_encode_turn, .parse_object = claude_parse_object, }; diff --git a/src/chat/chat-session.c b/src/chat/chat-session.c index 298d49f5..7a77d980 100644 --- a/src/chat/chat-session.c +++ b/src/chat/chat-session.c @@ -3,6 +3,8 @@ #include "settings/agent-secrets.h" #include "util/host-launch.h" +#include + #ifndef G_OS_WIN32 #include #endif @@ -23,12 +25,31 @@ struct _XdChatSession GSubprocess *process; GDataInputStream *stdout_stream; GDataInputStream *stderr_stream; + GOutputStream *stdin_stream; /* held open only by a streaming backend */ GCancellable *cancellable; GString *stderr_text; + /* + * What the running process was launched with. + * + * Model, effort and access are argv, so they are fixed for as long as the + * process lives. A turn that wants different ones cannot be handed to it and + * needs a new one -- which is why these are kept rather than the process + * simply being reused for anything. + */ + gboolean streaming; + char *session_id; /* last one the backend reported */ + char *launched_model; + char *launched_system_prompt; /* as given, before secrets were appended */ + char *launched_secrets; /* what the environment was built from */ + char *launched_workdir; + AiEffort launched_effort; + AiAccess launched_access; + guint kill_timeout_id; gboolean stopping; - gboolean finished; + gboolean turn_complete; /* the result line landed; finish after it */ + gboolean finished; /* of the current turn, not of the process */ }; enum @@ -48,6 +69,13 @@ G_DEFINE_FINAL_TYPE (XdChatSession, xd_chat_session, G_TYPE_OBJECT) static void read_next_line (XdChatSession *self); +static int +compare_strings (const void *a, + const void *b) +{ + return g_strcmp0 (*(const char *const *) a, *(const char *const *) b); +} + /* --- finishing ------------------------------------------------------------ */ static void @@ -82,8 +110,31 @@ on_process_waited (GObject *source, g_autoptr (XdChatSession) self = user_data; g_autoptr (GError) error = NULL; + /* + * A streaming process is not supposed to be gone. If a turn was still in + * flight it is now unanswerable and has to be reported; if none was, the + * chat simply has no process any more and the next turn starts one. + */ + if (self->streaming) + self->stdin_stream = NULL; + if (g_subprocess_wait_check_finish (G_SUBPROCESS (source), result, &error)) { + /* + * A streaming process exits only after its turn, never during one, so a + * clean exit with the turn still open means the CLI went away mid-answer. + * Calling that a success stores whatever partial text arrived as if it + * were the whole reply. + */ + if (self->streaming && !self->finished && !self->stopping) + { + const char *tail = stderr_tail (self); + + finish (self, FALSE, + tail != NULL ? tail : "The agent stopped before answering."); + return; + } + finish (self, TRUE, NULL); return; } @@ -120,6 +171,8 @@ on_event (const AiEvent *event, switch (event->type) { case AI_EVENT_SESSION_STARTED: + g_free (self->session_id); + self->session_id = g_strdup (event->session_id); g_signal_emit (self, signals[SIGNAL_SESSION_STARTED], 0, event->session_id); break; @@ -146,7 +199,24 @@ on_event (const AiEvent *event, case AI_EVENT_RESULT: /* The backend reported the id only at the end; keep it either way. */ if (event->session_id != NULL) - g_signal_emit (self, signals[SIGNAL_SESSION_STARTED], 0, event->session_id); + { + g_free (self->session_id); + self->session_id = g_strdup (event->session_id); + g_signal_emit (self, signals[SIGNAL_SESSION_STARTED], 0, event->session_id); + } + + /* + * For a process that exits when the turn does, the exit is the end and + * saying so here would be early -- stderr may still explain a failure. + * A streaming process outlives its turn, so this line is the only thing + * that says the turn is over. + * + * Noted rather than acted on: one line can carry several events, and + * whoever handles "finished" is entitled to drop the turn, which would + * leave the rest of the line being parsed into freed memory. + */ + if (self->streaming) + self->turn_complete = TRUE; break; case AI_EVENT_ERROR: @@ -192,7 +262,15 @@ on_line_read (GObject *source, ai_parser_feed_line (self->parser, line, on_event, self); + /* Reading continues first: the process is still there between turns, and + * whoever handles "finished" may start the next one straight away. */ read_next_line (self); + + if (self->turn_complete) + { + self->turn_complete = FALSE; + finish (self, TRUE, NULL); + } } static void @@ -226,6 +304,67 @@ on_stderr_line (GObject *source, /* --- lifecycle ------------------------------------------------------------ */ +/* + * What the secrets would put in an environment, as one comparable string. + * + * Names alone are not enough: a rotated value leaves the names identical while + * making the running process's environment wrong, and a process launched with + * the old value would keep using it for as long as the chat stayed open. The + * secrets are applied to an empty environment and hashed, so a changed value + * shows up the same way a changed name does. Sorted first, because the order + * they come out in is not part of what they are. + */ +static char * +secrets_fingerprint (XdAgentSecrets *secrets) +{ + g_auto (GStrv) applied = NULL; + g_autoptr (GChecksum) sum = g_checksum_new (G_CHECKSUM_SHA256); + + if (secrets == NULL) + return g_strdup (""); + + applied = xd_agent_secrets_apply_environment (secrets, g_new0 (char *, 1)); + if (applied == NULL) + return g_strdup (""); + + qsort (applied, g_strv_length (applied), sizeof (char *), compare_strings); + for (gsize i = 0; applied[i] != NULL; i++) + g_checksum_update (sum, (const guchar *) applied[i], -1); + + return g_strdup (g_checksum_get_string (sum)); +} + +/* + * Hands one turn to a process that reads them from stdin. + * + * Written synchronously: it is a single short line into a pipe with a whole + * turn about to follow it, and an async write here would only add a way for + * two turns to interleave on the same descriptor. + */ +static gboolean +write_turn (XdChatSession *self, + const AiRunSpec *spec, + GError **error) +{ + g_autofree char *line = self->backend->encode_turn (self->backend, spec); + g_autofree char *framed = NULL; + + if (line == NULL) + { + g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_FAILED, + "The turn could not be encoded."); + return FALSE; + } + + framed = g_strconcat (line, "\n", NULL); + + if (!g_output_stream_write_all (self->stdin_stream, framed, strlen (framed), + NULL, NULL, error)) + return FALSE; + + return g_output_stream_flush (self->stdin_stream, NULL, error); +} + XdChatSession * xd_chat_session_new (const AiBackend *backend) { @@ -258,6 +397,8 @@ xd_chat_session_start (XdChatSession *self, g_return_val_if_fail (spec != NULL, FALSE); g_return_val_if_fail (self->process == NULL, FALSE); + self->streaming = self->backend->encode_turn != NULL; + secrets = xd_agent_secrets_load (NULL, &local_error); if (secrets == NULL) { @@ -311,9 +452,26 @@ xd_chat_session_start (XdChatSession *self, return FALSE; } - /* codex reads stdin when it is a pipe and appends it to the prompt, so it - * has to see end-of-file straight away. */ - g_output_stream_close (g_subprocess_get_stdin_pipe (self->process), NULL, NULL); + if (self->streaming) + { + self->stdin_stream = g_subprocess_get_stdin_pipe (self->process); + + self->launched_model = g_strdup (effective.model); + /* The one that was asked for, not the one the secrets were appended to: + * it is what a later turn can be compared against. */ + self->launched_system_prompt = g_strdup (spec->system_prompt); + self->launched_secrets = secrets_fingerprint (secrets); + self->launched_workdir = g_strdup (effective.workdir); + self->launched_effort = effective.effort; + self->launched_access = effective.access; + } + else + { + /* codex reads stdin when it is a pipe and appends it to the prompt, so + * it has to see end-of-file straight away. */ + g_output_stream_close (g_subprocess_get_stdin_pipe (self->process), + NULL, NULL); + } self->stdout_stream = g_data_input_stream_new (g_subprocess_get_stdout_pipe (self->process)); @@ -326,6 +484,22 @@ xd_chat_session_start (XdChatSession *self, g_buffered_input_stream_set_buffer_size (G_BUFFERED_INPUT_STREAM (self->stdout_stream), 1 << 20); + /* + * The first turn goes out before anything is listening for the answer. + * Failing here means the process was never given the prompt, and returning + * with reads already queued would leave them to land on a caller that has + * been told the turn never started. + */ + if (self->streaming && !write_turn (self, &effective, error)) + { + g_subprocess_force_exit (self->process); + self->stdin_stream = NULL; + g_clear_object (&self->stdout_stream); + g_clear_object (&self->stderr_stream); + g_clear_object (&self->process); + return FALSE; + } + read_next_line (self); g_data_input_stream_read_line_async (self->stderr_stream, G_PRIORITY_LOW, @@ -335,6 +509,103 @@ xd_chat_session_start (XdChatSession *self, return TRUE; } +/* + * Whether a turn can be handed to the process that is already running. + * + * Everything compared here is fixed when the process starts and cannot be + * changed afterwards -- argv for most of it, the environment for the secrets. + * Someone who switches model mid-chat gets a new process, which is what they + * would have got before any of this. + * + * Instructions and secrets are in the list because they are the two that look + * like they took effect and did not: editing a folder's instructions or + * rotating a key would leave the running process on the old ones, silently, + * for as long as the chat stayed open. + */ +static gboolean +matches_launch (XdChatSession *self, + const AiRunSpec *spec, + const char *secrets) +{ + return g_strcmp0 (self->launched_model, spec->model) == 0 && + g_strcmp0 (self->launched_workdir, spec->workdir) == 0 && + g_strcmp0 (self->launched_system_prompt, spec->system_prompt) == 0 && + g_strcmp0 (self->launched_secrets, secrets) == 0 && + self->launched_effort == spec->effort && + self->launched_access == spec->access; +} + +gboolean +xd_chat_session_can_continue (XdChatSession *self, + const AiBackend *backend, + const AiRunSpec *spec) +{ + g_return_val_if_fail (XD_IS_CHAT_SESSION (self), FALSE); + g_return_val_if_fail (spec != NULL, FALSE); + + /* Switching a chat from claude to codex leaves a perfectly healthy process + * that is the wrong one to say this to. */ + if (backend != self->backend) + return FALSE; + + if (!self->streaming || self->process == NULL || + self->stdin_stream == NULL || self->stopping || !self->finished) + return FALSE; + + { + g_autoptr (XdAgentSecrets) secrets = xd_agent_secrets_load (NULL, NULL); + g_autofree char *fingerprint = secrets_fingerprint (secrets); + + return matches_launch (self, spec, fingerprint); + } +} + +gboolean +xd_chat_session_continue (XdChatSession *self, + const AiRunSpec *spec, + GError **error) +{ + g_autoptr (XdAgentSecrets) secrets = NULL; + g_autofree char *secret_prompt = NULL; + g_autofree char *system_prompt = NULL; + AiRunSpec effective; + + g_return_val_if_fail (XD_IS_CHAT_SESSION (self), FALSE); + g_return_val_if_fail (spec != NULL, FALSE); + + if (!xd_chat_session_can_continue (self, self->backend, spec)) + { + g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED, + "This turn needs a process of its own."); + return FALSE; + } + + /* The secrets prompt was appended to the system prompt in argv and is + * already in effect; only the turn's own text is new. */ + effective = *spec; + secrets = xd_agent_secrets_load (NULL, NULL); + if (secrets != NULL) + secret_prompt = xd_agent_secrets_prompt (secrets); + if (secret_prompt != NULL) + { + system_prompt = g_strdup (self->launched_system_prompt); + effective.system_prompt = system_prompt; + } + + self->finished = FALSE; + self->stopping = FALSE; + g_string_truncate (self->stderr_text, 0); + + /* The parser remembers what it has already handed out, so that a reply + * streamed as deltas and then repeated whole is not shown twice. That memory + * is about one turn; carrying it into the next would swallow the new one. */ + ai_parser_free (self->parser); + self->parser = ai_parser_new (self->backend); + ai_parser_set_model (self->parser, effective.model); + + return write_turn (self, &effective, error); +} + static gboolean on_grace_elapsed (gpointer user_data) { @@ -390,9 +661,15 @@ xd_chat_session_dispose (GObject *object) g_clear_handle_id (&self->kill_timeout_id, g_source_remove); - if (self->process != NULL && !self->finished) + /* + * A streaming process is idle between turns rather than gone, so "the turn + * finished" no longer means there is nothing to stop. Letting go of the + * session is what ends it, and not doing this left one behind per chat. + */ + if (self->process != NULL && (self->streaming || !self->finished)) g_subprocess_force_exit (self->process); + self->stdin_stream = NULL; g_cancellable_cancel (self->cancellable); g_clear_object (&self->stdout_stream); @@ -409,6 +686,11 @@ xd_chat_session_finalize (GObject *object) XdChatSession *self = XD_CHAT_SESSION (object); g_clear_pointer (&self->parser, ai_parser_free); + g_clear_pointer (&self->session_id, g_free); + g_clear_pointer (&self->launched_model, g_free); + g_clear_pointer (&self->launched_system_prompt, g_free); + g_clear_pointer (&self->launched_secrets, g_free); + g_clear_pointer (&self->launched_workdir, g_free); g_string_free (self->stderr_text, TRUE); G_OBJECT_CLASS (xd_chat_session_parent_class)->finalize (object); diff --git a/src/chat/chat-session.h b/src/chat/chat-session.h index f756dc95..9907e181 100644 --- a/src/chat/chat-session.h +++ b/src/chat/chat-session.h @@ -29,6 +29,31 @@ gboolean xd_chat_session_start (XdChatSession *self, const AiRunSpec *spec, GError **error); +/* + * Runs another turn on the process that is already up. + * + * A backend whose CLI takes its prompt in argv ends when the turn does, so + * there is nothing to continue and this refuses. Where it succeeds, the + * process was never restarted -- which is the point: anything the agent left + * running is still running, and there is no start-up cost between messages. + * + * Refuses a turn the running process cannot serve, notably one that changed + * model, effort, access or working directory, all of which are fixed in argv. + * Ask first with can_continue, and start a new session when it says no. + */ +/* + * @backend is passed rather than assumed: a session belongs to the CLI it was + * created for, and a caller keeping one per chat has to notice when the chat + * has been switched to another. Handing a claude turn to codex would otherwise + * look exactly like continuing. + */ +gboolean xd_chat_session_can_continue (XdChatSession *self, + const AiBackend *backend, + const AiRunSpec *spec); +gboolean xd_chat_session_continue (XdChatSession *self, + const AiRunSpec *spec, + GError **error); + /* Asks the child to stop, then insists if it does not. */ void xd_chat_session_cancel (XdChatSession *self); diff --git a/src/chat/chat-view.c b/src/chat/chat-view.c index 7b2742dc..bbdf2edb 100644 --- a/src/chat/chat-view.c +++ b/src/chat/chat-view.c @@ -158,6 +158,15 @@ struct _XdChatView GHashTable *turns; /* chat id -> Turn* */ + /* + * The agent process behind each chat, outliving the turns it answers. + * + * A turn ends and is discarded; the CLI that ran it can usually take the + * next one. Holding it here is what leaves the agent's own background work + * running between messages. chat id -> XdChatSession*. + */ + GHashTable *sessions; + GtkWidget *header; /* owned by the toolbar */ AdwWindowTitle *title; GtkStack *stack; @@ -2315,6 +2324,11 @@ turn_free (gpointer data) { Turn *turn = data; + /* The session can outlive the turn now, so it has to be told this listener + * is gone rather than being left pointing at freed memory. */ + if (turn->session != NULL) + g_signal_handlers_disconnect_by_data (turn->session, turn); + g_clear_object (&turn->session); g_clear_pointer (&turn->chat_id, g_free); g_clear_pointer (&turn->backend_id, g_free); @@ -2953,25 +2967,14 @@ start_turn (XdChatView *self, turn->segment = g_string_new (NULL); turn->items = g_ptr_array_new_with_free_func ((GDestroyNotify) turn_item_free); - turn->session = xd_chat_session_new (backend); + /* The session is chosen once the spec is built, since whether the chat's + * existing agent can take this turn depends on what the turn asks for. */ + /* Taken now rather than when the reply lands: the model can be changed * while the agent is still working, and what answered is whatever was * running when the turn started. */ turn->label = reply_title (chat); - g_signal_connect (turn->session, "session-started", - G_CALLBACK (on_session_started), turn); - g_signal_connect (turn->session, "commands", - G_CALLBACK (on_commands), turn); - g_signal_connect (turn->session, "text-delta", - G_CALLBACK (on_text_delta), turn); - g_signal_connect (turn->session, "tool-use", - G_CALLBACK (on_tool_use), turn); - g_signal_connect (turn->session, "usage", - G_CALLBACK (on_usage), turn); - g_signal_connect (turn->session, "finished", - G_CALLBACK (on_turn_finished), turn); - g_hash_table_insert (self->turns, g_strdup (chat->id), turn); set_working (self, TRUE); @@ -3011,14 +3014,56 @@ start_turn (XdChatView *self, spec.access = chat->plan ? AI_ACCESS_PLAN : ai_access_from_string (chat->access); - if (!xd_chat_session_start (turn->session, &spec, &error)) - { - append_row (self, XD_MESSAGE_ERROR, error->message); - xd_storage_append_message (self->storage, chat->id, "error", - error->message, NULL, NULL, NULL); - g_hash_table_remove (self->turns, chat->id); - set_working (self, FALSE); - } + /* + * The chat's agent answers again if it can. + * + * Keeping the process is what leaves whatever it started -- a background + * shell, a watch, a build -- still there for the next message. Anything the + * running one cannot serve, notably a changed model, effort, access, + * directory or backend, takes a new process and ends the old one. + */ + { + XdChatSession *pooled = g_hash_table_lookup (self->sessions, chat->id); + gboolean continuing = + pooled != NULL && xd_chat_session_can_continue (pooled, backend, &spec); + gboolean started; + + turn->session = continuing ? g_object_ref (pooled) + : xd_chat_session_new (backend); + + g_signal_connect (turn->session, "session-started", + G_CALLBACK (on_session_started), turn); + g_signal_connect (turn->session, "commands", + G_CALLBACK (on_commands), turn); + g_signal_connect (turn->session, "text-delta", + G_CALLBACK (on_text_delta), turn); + g_signal_connect (turn->session, "tool-use", + G_CALLBACK (on_tool_use), turn); + g_signal_connect (turn->session, "usage", + G_CALLBACK (on_usage), turn); + g_signal_connect (turn->session, "finished", + G_CALLBACK (on_turn_finished), turn); + + started = continuing + ? xd_chat_session_continue (turn->session, &spec, &error) + : xd_chat_session_start (turn->session, &spec, &error); + + if (!started) + { + append_row (self, XD_MESSAGE_ERROR, error->message); + xd_storage_append_message (self->storage, chat->id, "error", + error->message, NULL, NULL, NULL); + /* Whatever was pooled either failed us or was never usable. */ + g_hash_table_remove (self->sessions, chat->id); + g_hash_table_remove (self->turns, chat->id); + set_working (self, FALSE); + } + else if (!continuing) + { + g_hash_table_insert (self->sessions, g_strdup (chat->id), + g_object_ref (turn->session)); + } + } update_send_button (self); } @@ -4272,6 +4317,10 @@ forget_chat_sessions (XdChatView *self, activate_empty_transcript (self); remove_transcript_page (self, page); xd_terminal_panel_forget_chat (self->terminal, xd_node_get_chat_id (chat)); + + /* Nor an agent, which would otherwise sit waiting for a message that + * cannot come. */ + g_hash_table_remove (self->sessions, xd_node_get_chat_id (chat)); } static int @@ -5333,6 +5382,8 @@ xd_chat_view_dispose (GObject *object) xd_node_set_active (self->chat, FALSE); g_clear_object (&self->chat); g_clear_pointer (&self->turns, g_hash_table_unref); + /* Dropping a session ends its process, so emptying this stops them all. */ + g_clear_pointer (&self->sessions, g_hash_table_unref); g_queue_clear (&self->transcript_lru); g_clear_pointer (&self->transcript_pages, g_hash_table_unref); g_clear_pointer (&self->attachments, g_ptr_array_unref); @@ -5362,6 +5413,8 @@ xd_chat_view_init (XdChatView *self) GtkWidget *empty = adw_status_page_new (); self->turns = g_hash_table_new_full (g_str_hash, g_str_equal, g_free, turn_free); + self->sessions = g_hash_table_new_full (g_str_hash, g_str_equal, + g_free, g_object_unref); self->transcript_pages = g_hash_table_new_full ( g_str_hash, g_str_equal, NULL, (GDestroyNotify) transcript_page_free); g_queue_init (&self->transcript_lru); diff --git a/src/remote/server.c b/src/remote/server.c index 485fd219..67b5554b 100644 --- a/src/remote/server.c +++ b/src/remote/server.c @@ -56,6 +56,16 @@ struct _XdRemoteServer /* Turns in flight, by chat. One per chat is the rule, and this is what * enforces it. chat id -> XdDaemonTurn*. */ GHashTable *turns; + + /* + * The agent process behind each chat, outliving the turns it answers. + * + * A turn is finished and discarded; the CLI that ran it can usually take the + * next one, and keeping it is what leaves a background shell, a watch or a + * build still running for the message after this one. chat id -> + * XdChatSession*. + */ + GHashTable *sessions; gboolean quiescing; GTask *quiesce_task; @@ -1045,6 +1055,10 @@ handle_delete_chat (Connection *connection, return; } + /* The chat is gone, so the agent kept for its next turn is waiting for a + * message that will never come. */ + g_hash_table_remove (connection->server->sessions, chat_id); + { GHashTableIter iter; gpointer value; @@ -1952,6 +1966,7 @@ start_daemon_turn (XdRemoteServer *self, } turn = xd_daemon_turn_new (self->storage, self->root_path); + xd_daemon_turn_set_sessions (turn, self->sessions); running = g_new0 (Running, 1); running->server = self; @@ -1975,6 +1990,7 @@ start_daemon_turn (XdRemoteServer *self, g_hash_table_insert (self->turns, g_strdup (chat_id), turn); + /* Everyone watching sees the message arrive and the work start, including * the device that sent it -- one path, so every screen agrees. */ broadcast_event (self, "turn-started", chat_id, "label", @@ -3734,6 +3750,7 @@ xd_remote_server_dispose (GObject *object) if (self->storage != NULL) g_signal_handlers_disconnect_by_data (self->storage, self); g_clear_pointer (&self->turns, g_hash_table_unref); + g_clear_pointer (&self->sessions, g_hash_table_unref); g_clear_pointer (&self->command_sets, g_hash_table_unref); if (self->terminals != NULL) { @@ -3772,6 +3789,10 @@ xd_remote_server_init (XdRemoteServer *self) self->connections = g_ptr_array_new (); self->turns = g_hash_table_new_full (g_str_hash, g_str_equal, g_free, g_object_unref); + /* Dropping a session is what ends its process, so the table owning them is + * also what stops them: emptying it leaves nothing behind. */ + self->sessions = g_hash_table_new_full (g_str_hash, g_str_equal, + g_free, g_object_unref); self->command_sets = g_hash_table_new_full ( g_str_hash, g_str_equal, g_free, (GDestroyNotify) g_strfreev); self->terminals = g_hash_table_new_full (g_str_hash, g_str_equal, diff --git a/src/remote/turn.c b/src/remote/turn.c index 9d6ec8ec..b2c50c6a 100644 --- a/src/remote/turn.c +++ b/src/remote/turn.c @@ -33,6 +33,7 @@ struct _XdDaemonTurn char *root_path; XdChatSession *session; + GHashTable *sessions; /* unowned; the daemon outlives its turns */ char *chat_id; char *backend_id; char *model_id; @@ -640,15 +641,6 @@ xd_daemon_turn_start (XdDaemonTurn *self, self->text = g_string_new (NULL); self->segment = g_string_new (NULL); self->items = g_ptr_array_new_with_free_func ((GDestroyNotify) item_free); - self->session = xd_chat_session_new (backend); - - g_signal_connect (self->session, "session-started", - G_CALLBACK (on_session_started), self); - g_signal_connect (self->session, "commands", G_CALLBACK (on_commands), self); - g_signal_connect (self->session, "text-delta", G_CALLBACK (on_text_delta), self); - g_signal_connect (self->session, "tool-use", G_CALLBACK (on_tool_use), self); - g_signal_connect (self->session, "usage", G_CALLBACK (on_usage), self); - g_signal_connect (self->session, "finished", G_CALLBACK (on_finished), self); spec.prompt = full_prompt; spec.workdir = workdir; @@ -661,13 +653,54 @@ xd_daemon_turn_start (XdDaemonTurn *self, * overwriting it. */ spec.access = chat->plan ? AI_ACCESS_PLAN : ai_access_from_string (chat->access); - if (!xd_chat_session_start (self->session, &spec, error)) - { - xd_storage_append_message (self->storage, chat_id, "error", - (*error)->message, NULL, NULL, NULL); - g_clear_object (&self->session); - return FALSE; - } + /* + * A turn is a short-lived thing; the process answering it need not be. + * + * The chat's session is kept by the daemon and borrowed here, so a CLI that + * can take a second turn is given one instead of being restarted -- which is + * what leaves anything the agent started still running. Where it cannot, a + * new one takes its place in the pool and the old one ends. + */ + { + XdChatSession *pooled = self->sessions != NULL + ? g_hash_table_lookup (self->sessions, chat_id) : NULL; + gboolean continuing = + pooled != NULL && xd_chat_session_can_continue (pooled, backend, &spec); + gboolean started; + + self->session = continuing ? g_object_ref (pooled) + : xd_chat_session_new (backend); + + g_signal_connect (self->session, "session-started", + G_CALLBACK (on_session_started), self); + g_signal_connect (self->session, "commands", G_CALLBACK (on_commands), self); + g_signal_connect (self->session, "text-delta", G_CALLBACK (on_text_delta), self); + g_signal_connect (self->session, "tool-use", G_CALLBACK (on_tool_use), self); + g_signal_connect (self->session, "usage", G_CALLBACK (on_usage), self); + g_signal_connect (self->session, "finished", G_CALLBACK (on_finished), self); + + started = continuing + ? xd_chat_session_continue (self->session, &spec, error) + : xd_chat_session_start (self->session, &spec, error); + + if (!started) + { + xd_storage_append_message (self->storage, chat_id, "error", + (*error)->message, NULL, NULL, NULL); + + /* Whatever was pooled either failed us or was never usable. */ + if (self->sessions != NULL) + g_hash_table_remove (self->sessions, chat_id); + + g_signal_handlers_disconnect_by_data (self->session, self); + g_clear_object (&self->session); + return FALSE; + } + + if (!continuing && self->sessions != NULL) + g_hash_table_insert (self->sessions, g_strdup (chat_id), + g_object_ref (self->session)); + } start_diff_tracker (self); @@ -773,6 +806,15 @@ xd_daemon_turn_new (XdStorage *storage, return self; } +void +xd_daemon_turn_set_sessions (XdDaemonTurn *self, + GHashTable *sessions) +{ + g_return_if_fail (XD_IS_DAEMON_TURN (self)); + + self->sessions = sessions; +} + static void xd_daemon_turn_dispose (GObject *object) { diff --git a/src/remote/turn.h b/src/remote/turn.h index 485c0448..f7c9c8a8 100644 --- a/src/remote/turn.h +++ b/src/remote/turn.h @@ -33,6 +33,17 @@ G_DECLARE_FINAL_TYPE (XdDaemonTurn, xd_daemon_turn, XD, DAEMON_TURN, GObject) XdDaemonTurn *xd_daemon_turn_new (XdStorage *storage, const char *root_path); +/* + * Where the chat sessions live, keyed by chat id and owned by the caller. + * + * A turn ends and is thrown away; the process that answered it can be worth + * keeping, so it is held here instead and borrowed by whichever turn comes + * next. Unset, every turn gets a process of its own, which is what a turn + * created only to resolve a working directory wants. + */ +void xd_daemon_turn_set_sessions (XdDaemonTurn *self, + GHashTable *sessions); + /* * Starts @prompt as a turn in @chat_id. * diff --git a/tests/test-remote.c b/tests/test-remote.c index cc373b7c..f3fd186b 100644 --- a/tests/test-remote.c +++ b/tests/test-remote.c @@ -1560,6 +1560,10 @@ test_images_are_uploaded_to_the_daemon (void) g_assert_true (g_file_set_contents ( program, "#!/bin/sh\n" + /* The turn arrives on stdin now, so a stand-in waits for one the way the + * real CLI does. Exiting before it is read would leave xd writing a + * prompt into a pipe with nothing on the other end. */ + "read -r _turn || exit 1\n" "printf '%s\\n' " "'{\"type\":\"system\",\"subtype\":\"init\"," "\"session_id\":\"test-image-upload\"," @@ -2878,6 +2882,10 @@ test_send_during_turn_queues (void) g_assert_true (g_file_set_contents ( program, "#!/bin/sh\n" + /* The turn arrives on stdin now, so a stand-in waits for one the way the + * real CLI does. Exiting before it is read would leave xd writing a + * prompt into a pipe with nothing on the other end. */ + "read -r _turn || exit 1\n" "printf '%s\\n' " "'{\"type\":\"system\",\"subtype\":\"init\"," "\"session_id\":\"test-send-race\"}'\n" @@ -3082,6 +3090,10 @@ test_slow_git_snapshot_does_not_stall_other_chats (void) g_assert_true (g_file_set_contents ( claude_program, "#!/bin/sh\n" + /* The turn arrives on stdin now, so a stand-in waits for one the way the + * real CLI does. Exiting before it is read would leave xd writing a + * prompt into a pipe with nothing on the other end. */ + "read -r _turn || exit 1\n" "printf '%s\\n' " "'{\"type\":\"system\",\"subtype\":\"init\"," "\"session_id\":\"test-concurrent-chat\"}'\n" @@ -3187,6 +3199,10 @@ test_steer_starts_an_idle_remote_queue (void) g_assert_true (g_file_set_contents ( program, "#!/bin/sh\n" + /* The turn arrives on stdin now, so a stand-in waits for one the way the + * real CLI does. Exiting before it is read would leave xd writing a + * prompt into a pipe with nothing on the other end. */ + "read -r _turn || exit 1\n" "printf '%s\\n' " "'{\"type\":\"system\",\"subtype\":\"init\"," "\"session_id\":\"test-steered\"}'\n" @@ -3293,6 +3309,10 @@ test_a_joining_device_sees_an_active_turn (void) g_assert_true (g_file_set_contents ( program, "#!/bin/sh\n" + /* The turn arrives on stdin now, so a stand-in waits for one the way the + * real CLI does. Exiting before it is read would leave xd writing a + * prompt into a pipe with nothing on the other end. */ + "read -r _turn || exit 1\n" "printf '%s\\n' " "'{\"type\":\"system\",\"subtype\":\"init\"," "\"session_id\":\"test-running\"}'\n" @@ -3495,6 +3515,10 @@ test_a_live_turn_is_already_durable (void) g_assert_true (g_file_set_contents ( program, "#!/bin/sh\n" + /* The turn arrives on stdin now, so a stand-in waits for one the way the + * real CLI does. Exiting before it is read would leave xd writing a + * prompt into a pipe with nothing on the other end. */ + "read -r _turn || exit 1\n" "printf '%s\\n' " "'{\"type\":\"system\",\"subtype\":\"init\"," "\"session_id\":\"test-live-durable\"}'\n" @@ -3569,6 +3593,10 @@ test_a_restarted_daemon_resumes_interrupted_work (void) g_assert_true (g_file_set_contents ( program, "#!/bin/sh\n" + /* The turn arrives on stdin now, so a stand-in waits for one the way the + * real CLI does. Exiting before it is read would leave xd writing a + * prompt into a pipe with nothing on the other end. */ + "read -r _turn || exit 1\n" "printf '%s\\n' " "'{\"type\":\"system\",\"subtype\":\"init\"," "\"session_id\":\"test-update-resume\"}'\n" @@ -3725,6 +3753,10 @@ test_an_interrupted_turn_keeps_its_timeline (void) g_assert_true (g_file_set_contents ( program, "#!/bin/sh\n" + /* The turn arrives on stdin now, so a stand-in waits for one the way the + * real CLI does. Exiting before it is read would leave xd writing a + * prompt into a pipe with nothing on the other end. */ + "read -r _turn || exit 1\n" "printf '%s\\n' " "'{\"type\":\"system\",\"subtype\":\"init\"," "\"session_id\":\"test-interrupted\"}'\n" diff --git a/tests/test-session.c b/tests/test-session.c index 74f665b9..b864c498 100644 --- a/tests/test-session.c +++ b/tests/test-session.c @@ -54,6 +54,57 @@ static const AiBackend stub_backend = { .parse_object = stub_parse_object, }; +/* + * A CLI that takes its turns on stdin, the way claude does with + * --input-format stream-json: it answers each line and waits for the next + * rather than exiting, which is the whole point of keeping it. + * + * Launches are counted here rather than by the stub writing a file: the count + * is what tells one process answering twice from two processes answering once, + * and doing it in the test's own memory keeps a Windows path from having to + * survive being quoted into a shell script. + */ +static guint stream_launches; + +static GPtrArray * +stream_build_argv (const AiBackend *self, + const AiRunSpec *spec) +{ + GPtrArray *argv = g_ptr_array_new_with_free_func (g_free); + + /* Called once per spawn and never for a continued turn, so this counts + * processes without a file, a path, or a shell to quote one through. */ + stream_launches++; + + g_ptr_array_add (argv, g_strdup (self->program)); + g_ptr_array_add (argv, g_strdup ("-c")); + g_ptr_array_add (argv, g_strdup ( + "while IFS= read -r line; do " + " printf '%s\\n' '{\"type\":\"result\",\"subtype\":\"success\"," + "\"session_id\":\"kept\",\"is_error\":false,\"result\":\"ok\"}'; " + "done")); + g_ptr_array_add (argv, NULL); + + return argv; +} + +static char * +stream_encode_turn (const AiBackend *self, + const AiRunSpec *spec) +{ + return g_strdup_printf ("{\"prompt\":\"%s\"}", + spec->prompt != NULL ? spec->prompt : ""); +} + +static const AiBackend stream_backend = { + .id = "stream", + .display_name = "Stream", + .program = "sh", + .build_argv = stream_build_argv, + .encode_turn = stream_encode_turn, + .parse_object = stub_parse_object, +}; + static const AiBackend missing_backend = { .id = "missing", .display_name = "Missing", @@ -296,6 +347,65 @@ test_agent_secret_reaches_process_not_prompt (void) g_rmdir (directory); } + +/* + * The point of the whole change: a second turn reuses the first one's process. + * + * Counted by launches rather than by anything the session reports, because a + * session that quietly restarted would look identical from the outside and is + * exactly the failure worth catching. + */ +static void +test_reuses_the_process_for_a_second_turn (void) +{ + g_autoptr (XdChatSession) session = xd_chat_session_new (&stream_backend); + g_autoptr (GError) error = NULL; + AiRunSpec spec = { 0 }; + Run first = { 0 }; + Run second = { 0 }; + + guint watchdog; + + /* Each watchdog is taken down with its loop: one left armed fires into a + * Run that has already been cleared. */ + stream_launches = 0; + spec.prompt = "one"; + + run_init (&first, session); + g_assert_true (xd_chat_session_start (session, &spec, &error)); + g_assert_no_error (error); + watchdog = g_timeout_add_seconds (10, on_timeout, &first); + g_main_loop_run (first.loop); + g_clear_handle_id (&watchdog, g_source_remove); + g_assert_true (first.finished); + g_assert_true (first.success); + + /* The turn is over and the process is not: that is the difference. */ + g_assert_true (xd_chat_session_can_continue (session, &stream_backend, &spec)); + + /* The first run's handlers are still on the session, and its Run is about + * to stop being valid: one session outliving several turns is new, and so + * is having to take listeners off it. */ + g_signal_handlers_disconnect_by_data (session, &first); + run_clear (&first); + run_init (&second, session); + + spec.prompt = "two"; + g_assert_true (xd_chat_session_continue (session, &spec, &error)); + g_assert_no_error (error); + watchdog = g_timeout_add_seconds (10, on_timeout, &second); + g_main_loop_run (second.loop); + g_clear_handle_id (&watchdog, g_source_remove); + g_assert_true (second.finished); + g_assert_true (second.success); + + /* Two turns, one process. Two would mean it quietly restarted, which is + * indistinguishable from working if you only watch what it reports. */ + g_assert_cmpuint (stream_launches, ==, 1); + + run_clear (&second); +} + int main (int argc, char *argv[]) @@ -320,6 +430,8 @@ main (int argc, g_test_add_func ("/session/nonzero-exit", test_nonzero_exit_is_a_failure); g_test_add_func ("/session/agent-secret-environment", test_agent_secret_reaches_process_not_prompt); + g_test_add_func ("/session/reuses-the-process-for-a-second-turn", + test_reuses_the_process_for_a_second_turn); { int status = g_test_run ();