はじめに 前提：エージェントの構成 課題：画面を離れると回答が消える 設計方針：実行の作成と購読を分離する DB スキーマ：残す履歴と実行中の状態を分ける 保存単位は「AG-UI の Message 1 件 = 1 行」 DynamoDB ではなく Aurora MySQL を選んだ理由 会話履歴はサーバーが組み立てる 書き込み設計：イベントは保存せず「生成途中の回答」を上書きする 再合流：会話全体のスナップショットで追いつく 同時実行制御：実行中ロックを NULL 可のユニーク列で作る ロックの解放漏れに備える 停止：切断とキャンセルを区別する Redis は要るか まとめ はじめに 開発本部 開発1部の いくまる です。 私たちのチームでは、Web アプリの新機能として、チャット形式でデータを分析できる AI エージェントを開発中です。開発を進める中で、「回答の生成中に画面を離れると、その回答を受け取れなくなり、会話も残らない」という課題に向き合うことになりました。 本記事では、この課題を解決するために行った「会話履歴の永続化」と「バックグラウンド実行」の設計と実装を紹介します。DB スキーマ・実装コード・検討して捨てた案まで含めて書きます。 前提：エージェントの構成 このエージェントは次の構成で動いています。 ブラウザ（チャット UI） │ AG-UI イベント（SSE） ▼ Next.js（API Route ── ブラウザと AgentCore の中継役） │ InvokeAgentRuntime（SSE） ▼ Amazon Bedrock AgentCore Runtime（Strands Agents 製エージェント） │ MCP ▼ MCP サーバー（自社データの検索・集計ツール群） Amazon Bedrock AgentCore : AI エージェントの実行基盤となる AWS のサービスです。セッションごとに microVM 単位で実行環境が分離されます。 Strands Agents : AWS が公開しているオープンソースの AI エージェント SDK です。 AG-UI : エージェントとフロントエンド間のイベントストリーミングのプロトコルです。 RUN_STARTED ・ TEXT_MESSAGE_CONTENT ・ TOOL_CALL_* ・ RUN_FINISHED などのイベント型を定めています。転送方法は SSE に限定されませんが、このアプリでは SSE で流しています。 ユーザーが質問を送ると、エージェントが MCP ツールでデータを取得・分析し、回答をストリーミングで返します。ツールを繰り返し呼ぶため、1 回の回答に数十秒かかることがあります。 課題：画面を離れると回答が消える AI チャットで広く使われるのは「POST + ストリーミング応答」の構成です。ブラウザが質問を POST し、サーバーが生成イベントを流し、画面に逐次表示します。画面離脱を想定しなければ、これで十分に機能します。私たちの初期実装もこの構成でした。 私たちの場合は回答に数十秒かかるため、「生成中に画面を離れても実行は完走してほしい」という要件が加わりました。この要件を満たそうとすると、3 箇所が問題になります。 改修前の構造。画面遷移した瞬間に、以降のイベントを受け取る手段がなくなる 1 つ目はフロントエンドです。 この構成では、チャット画面の hook が unmount 時に実行を中断（ abortRun() ）する作りになりがちです。画面遷移がそのまま実行中断になります。 // チャット画面の hook（unmount 時に実行を中断する作り） useEffect( () => () => { stopRequestedRef. current = true ; agentRef. current ?.abortRun(); } , [] , ); 2 つ目は中継役の API Route です。 ブラウザと AgentCore の間で SSE を中継する Next.js の API Route は、クライアントの切断を上流の AgentCore への読み取りキャンセルとして伝播します。 ReadableStream の cancel() は「クライアントがもう読まない」ときに呼ばれるコールバックで、そこで上流の読み取りも止めると、切断とキャンセルの区別がなくなります。 // 中継処理（クライアントが切れると上流の読み取りも止まる作り） const readable = new ReadableStream( { async start ( controller ) { // 上流（AgentCore）の SSE を読み、そのままクライアントへ中継する } , cancel () { reader.cancel(); } , } ); 3 つ目は保存先です。 イベントは中継されるだけで、どこにも保存されません。仮に 1 つ目と 2 つ目を直して実行が完走するようにしても、戻ってきた画面に表示するデータがありません。 「実行状態を React のグローバルストアに持てば、画面遷移に耐えられるのでは」という案も検討しました。しかしこのアプリでは、チャット画面から他の画面への遷移が window.location.href によるページ全体の再読み込みで実装されています。再読み込み後のページは JavaScript の実行環境ごと新しく作られるため、React の state やグローバルストアに入れた値は引き継がれません。 設計方針：実行の作成と購読を分離する 大きく変えたのは次の 3 つです。 作成と購読の分離 : POST /runs は実行を開始して 202 { runId } を即座に返します。表示は GET /runs/{runId}/events の SSE で受け取ります。この「SSE を受信し続けること」を、本記事では「購読」と呼びます。購読はいつ切れてもよく、何度でも再開できます。 実行ワーカーの独立 : AgentCore の SSE を最後まで読み切って記録する処理（実行ワーカー）を、HTTP レスポンスから独立した非同期タスクにしました。ブラウザが切断しても実行は完走します。 二層の保存 : 実行中は「生成途中の回答」を DB に上書き保存し続け、完了したら完成したメッセージを DB の履歴テーブルに保存します。テーブル構成は次の節で説明します。 なお、実行ワーカーは Next.js と同じプロセス内で動かしているため、リクエストごとに実行環境が終了するサーバーレス環境ではこの形は取れません。現在は検証段階のためこの構成にしていますが、Next.js のデプロイに走行中の実行が巻き込まれないようにするため、本来は実行ワーカーを独立したプロセスに切り出す方が望ましいと考えています。現状、プロセスがデプロイなどで止まる場合の後始末は、同時実行制御の節で説明する回収の仕組みが担います。 改修後の構造。実行は接続と無関係に完走し、購読は何度でも再入場できる ブラウザとサーバーの間の API は次の 5 本です。 API 役割 POST /runs 実行を作成して 202 { runId, conversationId } を即返す GET /runs/{id}/events SSE 購読。切断・再入場が自由 POST /runs/{id}/cancel 明示的なキャンセル GET /conversations 会話一覧（履歴サイドバー用） GET /conversations/{id} 会話の全メッセージ + 実行中の run（あれば） 最後の API がポイントです。リロードや別タブで会話を開いた直後、クライアントは会話 ID しか知らず、実行中の run があるかどうかも分かりません。そこで GET /conversations/{id} は、会話のメッセージに加えて「実行中の run の ID」を返します。クライアントはその ID で GET /runs/{id}/events を購読し、生成途中から表示を再開します。 DB スキーマ：残す履歴と実行中の状態を分ける このアプリでは以前から、本体機能のデータを Aurora MySQL 8.0 + Prisma で管理しています。エージェントの履歴も同じ DB に、3 つのテーブルで持つことにしました。ずっと残す「履歴」と、実行中だけ使う「実行状態」でテーブルを分けています。 区分 テーブル 役割 行の扱い 履歴 conversations 会話スレッド 1 件のメタ情報 ずっと残す 履歴 conversation_messages メッセージ 1 件 = 1 行。完成した発話を保存 ずっと残す 実行状態 runs 実行 1 回の状態 + 生成途中の回答 + ロック 行は実行 1 回ごとに増え、終了後も記録として残る。生成途中の回答やロックは実行中だけ使う ER 図 Prisma スキーマは次の通りです。実際に採用したものから、タイムスタンプ列・リレーション定義・enum 定義（RunStatus / MessageStatus）を省いています。 model Conversation { id String @id @default(cuid()) companyId Int @map("company_id") userId String @map("user_id") threadId String @map("thread_id") @db.Char(36) title String @db.VarChar(255) deletedAt DateTime? @map("deleted_at") @@unique([companyId, userId, threadId]) @@map("conversations") } model ConversationMessage { id String @id @default(cuid()) conversationId String @map("conversation_id") runId String? @map("run_id") sequence Int role String @db.VarChar(16) parts Json status MessageStatus @default(complete) @@unique([conversationId, sequence]) @@map("conversation_messages") } model Run { id String @id @default(cuid()) conversationId String @map("conversation_id") clientTurnId String @map("client_turn_id") @db.VarChar(64) status RunStatus @default(queued) lockKey String? @unique @map("lock_key") ownerInstanceId String? @map("owner_instance_id") @db.VarChar(64) errorCode String? @map("error_code") @db.VarChar(64) partialState Json? @map("partial_state") heartbeatAt DateTime @default(now()) @map("heartbeat_at") @@unique([conversationId, clientTurnId]) @@index([status, heartbeatAt]) @@map("runs") } lockKey の UNIQUE、 clientTurnId の複合ユニーク、 [status, heartbeatAt] のインデックスがそれぞれ何のためにあるかは、後の節で順に説明します。 また、本記事には 4 種類の ID が登場します。ここで整理しておきます。 ID 何を指すか conversationId DB 上の会話。API で会話を指すときに使う threadId AG-UI 上の会話 ID。DB 上の会話（ conversationId ）と 1 対 1 で対応 runId 質問 1 回ぶんの実行 runtimeSessionId AgentCore の実行環境を束ねる ID。 t{companyId}-u{userId}-{threadId} の形式で、会話ごとに固定 保存単位は「AG-UI の Message 1 件 = 1 行」 conversation_messages は追記専用で、AG-UI の Message をそのまま parts （JSON）に格納します。テキストだけのターンは user / assistant の 2 行、ツールを使うターンは assistant（ツール呼び出し）と tool（結果）の行が挟まって 4 行以上になります。 1 会話のメッセージ行の例。ツールを使うターンは user・assistant（ツール呼び出し）・tool・assistant の 4 行、使わないターンは 2 行になる この保存単位を選んだ理由は、フロントの表示ロジックの作りにあります。ライブ表示は「AG-UI の Message 配列を受け取り、ターンの区切りやツールの実行ステップ表示を組み立てる純粋関数」として自前で実装しています。保存した Message 列をそのままこの関数に渡せば、画面を離れなかった場合と同一の表示が再現されます。保存時に表示用の形へ加工してしまうと、同じ表示を再現できなくなります。 DynamoDB ではなく Aurora MySQL を選んだ理由 会話履歴のアクセスパターン（会話 ID + 連番で順に全件取得、追記専用、JSON 主体）は DynamoDB の得意領域で、実際に移行案も検討しました。それでも Aurora MySQL 一本にしています。 まず、トランザクション要件が構成の選択肢を絞ります。質問の受付時には「会話 + user メッセージ + 実行（ロック）の INSERT」を、完了時には「assistant メッセージの INSERT + 実行の完了 + ロック解放」を、それぞれ単一トランザクションで行う必要があります。「履歴は DynamoDB、実行状態は Aurora」のように 2 つのストアに分けると、この原子性を保証できません。原子性が無いと、たとえば次のような壊れ方をします。 回答は残ったのに次の質問ができない : 完了処理の「回答を保存」と「ロック解放」の間でプロセスが落ちると、画面には回答が出ているのに DB はロックを握ったままになり、次の質問が「実行中です」と拒否され続けます。 答えのない質問が履歴に残る : 送信の二度押しで 2 本目が「質問を保存 → ロックで弾かれる」の順に進むと、誰も回答しない質問だけが履歴に残ります。1 トランザクションならロック取得の失敗と同時に質問の保存も取り消され、エラー応答だけを返せます。 したがって選択肢は「全部 Aurora」か「全部 DynamoDB」に絞られます。後者も技術的には成立します（DynamoDB でも TransactWriteItems で複数の項目をまとめて原子的に書けます）。 それでも Aurora にしたのは、既存の運用との一貫性のためです。このアプリの他のデータはすべて Aurora + Prisma で管理していて、マイグレーションの手順やレビューの観点といったチームの運用もそこで揃っています。データストアを 2 つにすると、この運用も 2 系統になります。規模の面でも、DB への書き込みはピークでも毎秒数十回の見積もりで、Aurora で十分に受けられます。DynamoDB のスケール性能が必要になる水準ではありません。 会話履歴はサーバーが組み立てる エージェントは毎回の呼び出しで会話の全履歴を受け取り、状態をゼロから組み立て直す作りにしています。この全履歴を誰が用意するかには 2 つの形があります。クライアントが手元の Message 配列を毎回送るか、 サーバーが DB から組み立てる かです。私たちは後者にしました。クライアントが送るのは新しいメッセージ 1 件だけです。 前者を避けた理由は、バックグラウンド実行と相性が悪いからです。実行を放置して別のタブで完走させると、元のタブが持っている履歴は古いままになります。その古いタブから次の質問を履歴ごと送ると、完走したはずの回答がモデルへの入力から抜け落ち、会話のつじつまが合わなくなります。後述するロックは実行中しか効かないため、この事故は防げません。最新の会話を常に持っているのは DB だけです。 なお、AgentCore 側に会話の状態を持たせる案も 2 つ検討し、見送りました。 実行環境（microVM）のメモリに持つ : セッション ID は会話ごとに固定なので、同じ会話の呼び出しは同じ実行環境に届き、メモリに状態を残すこと自体はできます。ただしこの環境は無操作 15 分（デフォルト）などで終了し、メモリごと消えます。時間を空けて続く会話の置き場にはできません Memory サービスに持つ : AgentCore には会話を保存する Memory というサービスもあります。ただし履歴はどのみち表示のために自前の DB へ保存するので、足すと同じ役割の保存先が 2 つになります POST /runs のボディは { conversationId, message, clientTurnId } だけです。モデルへ渡す履歴は、実行ワーカーが Aurora から組み立てます。 // 実行ワーカーの一部：DB から会話履歴を読み出し、モデル入力用に整える export async function buildModelMessages ( conversationId : string ): Promise < ModelMessagesResult > { const rows = await prisma.conversationMessage.findMany( { where : { conversationId , status : "complete" } , orderBy : { sequence : "asc" } , select : { parts : true } , } ); return normalizeHistory(rows); } // 実行ワーカーの一部：組み立てた履歴をエージェントに渡して生成を開始する const history = await buildModelMessages(claimed.conversationId); const { companyId , userId , threadId } = claimed.conversation; const agent = new AgentCoreAgent( { threadId , initialMessages : history .messages, agentArn , runtimeUserId : `c ${ companyId } :u ${ userId } ` , runtimeSessionId : `t ${ companyId } -u ${ userId } - ${ threadId } ` , } ); こうすると、会話の内容はサーバー（DB）だけが持つ構造になります。AgentCore の microVM がタイムアウトで終了しても、クライアントがリロードで状態を失っても、会話は Aurora から再構成できます。 書き込み設計：イベントは保存せず「生成途中の回答」を上書きする 1 回の回答で AG-UI イベントは数十〜数百個流れます。本文の断片（デルタ）1 つ 1 つやツール呼び出しがそれぞれイベントになるため、回答が長いほど増えます。これを 1 行ずつ INSERT すると、1 回答ごとに大量の行が永久に積み上がります。採用したのは次の形です。 時点 DB 操作 内容 質問送信 INSERT conversations に会話（新規会話のときのみ）、 conversation_messages に質問、 runs に実行レコード（ロック取得を兼ねる）の最大 3 行 生成中 runs.partial_state を上書き UPDATE AG-UI のイベントでは本文が細切れ（デルタ）で届く。実行ワーカーはそれをつなぎ合わせた「その時点のメッセージ配列」を保持しており、これを数秒ごとに同じ 1 行へ上書き保存。行は増えない 完了 conversation_messages に INSERT そのターンで生まれたメッセージ（回答、ツールを使った場合はその呼び出しと結果も）を保存。 runs.partial_state を空にし、ロックを解放 質問送信時のトランザクションは以下のようになっています。ロック（ lockKey ）の取得と質問の保存が、同時に成立するか同時に失敗するかのどちらかになります。 // API Route の一部：質問受付時の書き込み return prisma.$transaction( async ( tx ) => { const conversationId = existingId ?? ( await tx.conversation.create( { /* 省略 */ } )).id; // ロック取得に失敗したら、下で保存する質問ごと取り消される（同一トランザクションのため） const run = await tx.run.create( { data : { conversationId , clientTurnId , lockKey : conversationId } , select : { id : true } , } ); // aggregate（集計クエリ）で会話内の最大 sequence を取り、次の連番を振る const highest = await tx.conversationMessage.aggregate( { where : { conversationId } , _max : { sequence : true } , } ); await tx.conversationMessage.create( { data : { conversationId , runId : run. id , sequence : (highest._max.sequence ?? 0 ) + 1 , role : "user" , parts : { id : randomUUID(), role : "user" , content : message } , } , select : { id : true } , } ); return { runId : run. id , conversationId } ; } ); 生成中の partial_state は 3 秒間隔で間引いて書きます。間引きに加えて「前の UPDATE が完了するまで、次の UPDATE を発行しない」という制御も入れています。UPDATE を発行した順と DB に反映される順は一致するとは限らないため、古い内容の UPDATE が新しい内容の後に適用されると、保存済みの「生成途中の回答」が巻き戻ってしまうからです。 // 実行ワーカーの一部：生成途中の回答を数秒ごとに DB へ上書き保存する const PARTIAL_STATE_INTERVAL_MS = 3_000 ; return { schedule() { // 前の書き込みが完了するまで次をスケジュールしない（古い内容への巻き戻りを防ぐ） if (timer || writing || truncated) return ; timer = setTimeout (() => { timer = null ; writing = true ; void writePartialState(runId, ownerInstanceId, produced()) . then (( state ) => { truncated = state.truncated; } ) . catch (( error : unknown ) => console .error( "partial_state の更新に失敗しました" , { runId , error } ), ) . finally (() => { writing = false ; } ); } , PARTIAL_STATE_INTERVAL_MS); } , } ; 数秒おきに UPDATE を発行し続けて DB の負荷は大丈夫なのか、という点は検討しました。結論としては、同時に走る生成が多くても数十本という規模では問題になりません。更新は各実行が自分の 1 行だけを主キー指定で行い、実行間のロック競合はありません。さらに、接続中のユーザーの画面へは実行ワーカーがメモリ上のイベントを直接流すため、 partial_state の用途は後述する再合流だけです。毎秒書く必要も、イベントを 1 個ずつ書く必要もありません。 完了時は「回答の確定保存」「実行ステータスの完了への更新」「ロック解放」「生成途中の回答（ partial_state ）の削除」を 1 トランザクションで行います。 // 実行ワーカーの一部：完了時の書き込み。lock_key を外し損ねるとその会話が永久に 409 になる return prisma.$transaction( async ( tx ) => { const claimed = await tx.run.updateMany( { where : terminableWhere(runId, ownerInstanceId), data : { status , errorCode , finishedAt : new Date (), lockKey : null , partialState : Prisma.DbNull, } , } ); // 別の経路（キャンセルや、後述する異常終了時の回収処理）が先にこの実行を // 終わらせていたら、生成物は保存しない if (claimed. count === 0 ) return false ; await tx.conversationMessage.createMany( { data : messages. map (( message , index ) => ( { conversationId : targetId, runId , sequence : base + index + 1 , role : message.role, parts : message as Prisma.InputJsonValue , status : messageStatusAt( status , message. id , openMessageIds), } )), } ); return true ; } ); メッセージの status は通常 complete で保存します。キャンセルやエラーで実行が正常に終わらなかった場合は、そこまでに生成できていた分を partial（部分的、の意味）として保存し、画面に残せるようにしています。 再合流：会話全体のスナップショットで追いつく 実行中の会話に購読者が入ってくると、サーバーはまず RUN_STARTED （「実行が進行中です」の合図）と MESSAGES_SNAPSHOT を送ります。どちらも上流から届いたイベントの中継ではなく、この購読のためにサーバーが新しく作って送るものです。 MESSAGES_SNAPSHOT を受け取ったクライアントは、手元のメッセージ一覧を捨てて、スナップショットの内容で丸ごと置き換えます。そのため、スナップショットに生成途中の 1 件だけを入れると、過去のメッセージが画面からすべて消えてしまいます。必ず会話の全メッセージを入れて送ります。 その後の配信は 2 つのモードに分かれます。分かれ目は「購読がいつ始まったか」です。 モード いつ使われるか 配信内容 live 質問の送信直後から購読している場合（生成イベントがまだ 1 件も流れていないうちに購読が始まったとき） 実行ワーカーが受け取る生成イベント（ TEXT_MESSAGE_CONTENT など）を、メモリからそのまま逐次中継 poll それ以外すべて（リロード・別タブ・離脱して戻ってきた場合） 1 秒間隔で DB を読み、確定済み履歴と partial_state の生成途中回答をマージした会話全体の MESSAGES_SNAPSHOT を、内容が変わったときだけ送り直す。実行が終わったら RUN_FINISHED （または RUN_ERROR ）で締める つまり、途中から戻ってきた購読者が受け取るのは live 配信のイベント列ではなく、「会話全体のスナップショットの送り直し」です。 // 配信モードの選択。イベントが 1 件でも流れた後に始まった購読は poll に回す export function attach ( runId : string , signal : AbortSignal ): LiveSubscription | null { const fanout = fanouts. get (runId); if (!fanout) return pollInstead(runId, "fanout_absent" ); if (fanout. closed ) return pollInstead(runId, "fanout_closed" ); // 1 件でも中継済みなら列の途中からになるので、履歴を出せる poll に任せる if (fanout.relayed > 0 ) return pollInstead(runId, "already_relayed" ); // 省略（購読者を登録し、生成イベントを流す AsyncGenerator を返す） } // poll 配信のループ。会話全体のスナップショットを、内容が変わったときだけ送り直す while ( true ) { const event = snapshot(stored, readPartialMessages(progress.partialState)); const serialized = JSON . stringify (event); if (serialized !== previous) { previous = serialized; yield event; } if (isTerminal(progress. status )) break ; await sleep(POLL_INTERVAL_MS, signal); progress = await readProgress(run. id ); if (isTerminal(progress. status )) stored = await loadConversationMessages(run.conversationId); } yield terminalEvent(run, progress); 途中合流の購読者を live のイベント列に合流させず poll に回すのは、正しさを優先したためです。デルタの続きから流すには、「スナップショットに含めた分」と「これから流すデルタ」の境界を厳密に合わせる必要があります。境界がずれると、 content += delta の積み上げで本文が二重に連結されます。会話全体のスナップショットを送り直す形なら、毎回が丸ごとの置き換えなので、この事故が原理的に起きません。その代わり、poll 配信の画面は live 配信のようなストリーミング表示にはならず、数秒おきに文章がまとまって進む表示になります。途中合流でもストリーミング表示にすることは、後続の課題にしています。 同時実行制御：実行中ロックを NULL 可のユニーク列で作る 「同一会話に実行中の run は 1 本だけ」を DB で強制します。PostgreSQL なら 部分インデックス （partial index。 CREATE UNIQUE INDEX ... WHERE status IN ('queued','running') のように、条件を満たす行だけへ一意制約をかけられます）で書けますが、MySQL 8.0 には相当する構文が用意されていません。 代わりに runs.lock_key （NULL 可・UNIQUE）を使いました。実行中は lock_key = conversationId 、終了時に NULL へ戻します。MySQL のユニークインデックスは NULL を重複として扱わない ため、終了済みの run は何本でも共存でき、実行中は会話ごとに 1 本に絞られます。 同一会話への 2 本目の POST /runs は、INSERT 時にユニーク制約違反のエラーとして原子的に弾かれます。重複には 2 種類あります。1 つは「同じ送信の二度押し」です。クライアントは送信 1 回ごとに ID（ client_turn_id ）を発行し、リトライでも同じ ID を送るため、この列の重複で検出できます。送信ボタンの連打はフロントでも抑止できますが、ネットワーク不調時の自動再送などフロントの制御では防げない経路が残るため、DB でも守ります。もう 1 つは「別の質問の並行送信」（ lock_key の重複）で、同じ会話を複数のタブで開いているときに起きます。どちらだったかを引き直して、応答を分岐します。 // API Route の一部：重複キーエラーの解釈 } catch (error) { if (!isUniqueViolation(error)) throw error; // 二度押しなら、先行の run をそのまま返す（同じ送信は 1 回として扱う） const raced = await findRunByClientTurnId(params); if (raced) return { ok : true , runId : raced. id , conversationId : raced.conversationId } ; // 並行送信なら、実行中の run を添えて拒否へ const activeRunId = await findActiveRunIn(conversationId); if (activeRunId) return { ok : false , reason : "active_run" , activeRunId } ; } // API Route の一部：並行送信への応答 if (result.reason === "active_run" ) { return Response .json( { errors : "この会話はいま実行中です" , activeRunId : result.activeRunId } , { status : 409 } , ); } 409 のレスポンスに activeRunId を含めているのは、UI がそれを使って「拒否」ではなく「実行中の run への購読切り替え」に変換できるようにするためです。 ロックの解放漏れに備える ロックには解放漏れへの備えも必要です。サーバーのプロセスが突然落ちると、running のままロックを握った run が残ります。そうなると、誰も実行していないのにその会話への質問が「実行中です」と拒否され続けます。備えは 2 つの仕組みの組み合わせです。 実行ワーカーは、実行中の run の heartbeat_at を定期的に現在時刻へ更新します（処理が続いていることの記録です） それとは別の掃除処理が、 heartbeat_at の更新が一定時間止まっている queued / running の run を「担当プロセスが異常終了した」とみなして failed にし、ロックを解放します。 heartbeat_at は INSERT 時に現在時刻が入るため、202 を返した直後・実行が始まる前にプロセスが落ちて queued のまま残った run も、この経路で回収されます（スキーマの @@index([status, heartbeatAt]) はこの検索用です） 「サーバー起動時に、残っている running を全部 failed にする」というより単純な方法は採れませんでした。デプロイ中は新旧のサーバーがしばらく同時に動いており、旧サーバーがまだ実行している最中の run を、新サーバーの起動処理が誤って failed にしてしまうためです。run に owner_instance_id （どのサーバーがその実行を担当しているか）を持たせているのも同じ理由です。掃除処理は、自分のサーバーがいま実行している run を誤って回収しないよう、この ID とメモリ上の実行一覧を突き合わせて判定します。 この回収の仕組みは、デプロイやスケールインでプロセスごと止められた場合の後始末も兼ねています。止まったプロセスが抱えていた実行は途中から再開できませんが、heartbeat が途絶えるため数分以内に failed になり、そこまでの生成分は partial として履歴に残り、会話のロックも解放されます。ユーザーは失敗を確認して、すぐ次の質問に進めます。 停止：切断とキャンセルを区別する この設計では、タブを閉じる・画面を遷移するのは「切断」であり、実行は継続します。明示的に止めたいときは POST /runs/{id}/cancel を呼びます。実行ワーカーにキャンセル要求の印を立てて上流への購読を切り離し、実行を「キャンセル」として記録します。記録後に遅れて届いた生成物は、前述の完了時トランザクションの「別の経路が先に終わらせていたら保存しない」分岐で破棄されます。 // API Route の一部：キャンセル処理 if (getActiveRun(runId) || run.ownerInstanceId === OWNER_INSTANCE_ID) { // 終了の記録は実行ワーカーに任せる。ここで書くと、ワーカーが「先に終了済み」と判定して生成物を捨てる const active = registerRun(runId); active.cancelRequested = true ; await active.agent?.detachActiveRun(); return { ok : true } ; } // 別のサーバーが担当している run。会話のロックを解放するために記録だけ書く await finalizeRun( { runId , status : "cancelled" , finalizedBy : `cancel: ${ OWNER_INSTANCE_ID } ` } ); 実行中の会話への追加送信は、前述のロックにより 409 で拒否されます。ただし 409 を返すだけだと、「戻ってきたら画面が止まって見える → もう一度送る → エラー」という流れになりやすいため、途中経過の可視化（再合流）を初回リリースの範囲に含めています。実行中であることが見えていれば追加送信は起きにくく、方向を変えたい場合も「停止してから送る」導線に誘導できます。 Redis は要るか 同種の設計では、実行中イベントの共有に Redis（Redis Streams）を使う構成がよく知られています。ただし Redis が必要になるのは「タスクを 2 つ以上に増やし、かつストリーミング表示を保ちたい」場合です。今回はどちらにも当てはまらないため、入れていません。 現在は ECS 1 タスクで動かしており、通常時は再接続のリクエストが実行ワーカーと同じプロセスに届きます。イベントはプロセス内のメモリで手渡せるため、Redis なしで live 配信が成立します。 タスクを 2 つ以上に増やすと、購読のリクエストが実行ワーカーのいない方のタスクへ届くことがあります。live 配信はワーカーと同じプロセスのメモリを介して成り立っているため、別のタスクに届いた購読では使えません。ただし DB はどのタスクからも読めるので、poll 配信はそのまま動きます。つまり live 配信できたはずの購読が poll 配信になり、ストリーミング表示が数秒おきの更新になるだけで、履歴も途中経過も見られます。設計方針の節で触れた「実行ワーカーを独立したプロセスに切り出す」場合も、ワーカーと購読者が必ず別プロセスになるため、同じく中継が必要になります。当面は 1 タスクで足りる規模のため、現時点ではこの構成にしています。 まとめ 実行の作成と購読を分離し、実行ワーカーを HTTP 接続から独立させることで、画面を離れても実行が完走する構造にしました。 履歴は「メッセージ 1 件 = 1 行」でずっと残し、生成途中の回答は上書き更新の 1 行に分けました。イベントの逐次保存はせず、モデルへ渡す履歴もサーバーが DB から組み立てます。 同時実行制御は MySQL の「NULL 可ユニーク列」によるロックで実現しました。キャンセルは購読の切り離しと、実行を「キャンセル」として記録することで実現し、切断とは明確に区別しています。 AI チャットの「履歴」と「バックグラウンド実行」は別々の機能に見えますが、作ってみると、どちらも「会話の状態はサーバー側で持つ」という同じ設計に行き着きました。AI チャットの実行基盤を作る際に共通して現れる論点だと思うので、同じものを作る方の参考になれば幸いです。

ミイダステックチームです！ ミイダス開発部では、定期的にエンジニアのスキルアップや最新技術の知見共有を目的とした社内勉強会を開催しています。

こんにちは。『楽楽請求』でフロントエンドを担当しているtakenamiです。 『楽楽請求』では立ち上げ当初から、要件定義から画面仕様の作成までを設計チームが担い、開発チームがそれを実装するという分担で開発を進めてきました。2024年10月のリリースから約1年半が経った2026年4月、顧客への価値提供スピードをさらに高めるための部の方針として、 UI設計をフロントエンドが担う体制 へと移行しています。 実際に担ってみると、実装を担当していた頃には見えていなかった景色と、いくつもの壁にぶつかりました。この記事では、この体制に至った背景・進め方と、現時点で向き合っている課題をお伝えします。 1. 前提となる開発体制 2. なぜ踏み出す必要があったのか 設計フェーズに負荷が集中していた なぜフロントエンドだったのか プロトタイプを作るコストが下がった プロダクトのフェーズも変わってきた 3. まずは小さくUI設計を担う 立ち上げは小さく、定着は着実に Before → After 進め方 4. 担ってみて見えてきた、3つの課題 課題① 顧客・業務理解を深める 課題② UIパターンの引き出しを増やす 課題③ 設計意図を言語化する 5. 一歩踏み出した先に見えてきた、次の景色 おわりに 1. 前提となる開発体制 『楽楽請求』は、クラウド型請求書処理システム市場へ後発として参入したプロダクトです。初期フェーズでは、市場のニーズに迅速に応え、PMF（Product Market Fit）を達成することが最重要課題でした。初期メンバーによる徹底した現場視点と迅速な価値提供があったからこそ、現在の『楽楽請求』の成長につながっています。 当時の役割分担は次のとおりです。 製品企画チーム ：どの機能を作るかの企画を担当 設計チーム ：要件定義から概要設計まで。その一環として、FigmaでのUI設計までを担当 開発チーム（バックエンド・フロントエンド） ：詳細設計・実装・テストを担当 この明確な分担は、開発効率を高めるうえで大きく機能してきました。設計チームは要件と画面仕様の検討に、開発チームは実装品質にそれぞれ集中できる。短期間で多くの機能を届けられてきたのは、この体制があったからだと思っています。 私が参画した当時、『楽楽請求』では楽楽シリーズ共通のUIの統一がすでに完了していました。統一された画面をベースにできるため、Figmaの画面仕様は設計チームが作成し、必要に応じてデザイナーに依頼する形で運用されています。 これから紹介するのは、この分担を否定する話ではありません。 共通基盤が整っているからこそ、UI設計を誰が担うのが最も速いかを、組織として問い直した 話です。 2. なぜ踏み出す必要があったのか 設計フェーズに負荷が集中していた 体制上の制約がありました。当時の設計チームは少人数で、事業部の製品企画と連携しながら、要件定義から画面仕様の作成までを一手に担っていたことです。 企画・要件・UI設計が直列でつながっているため、どれか一つが詰まれば後続がすべて待つ構造になります。開発チームは実装の準備ができていても、画面仕様が出てくるまで着手できない。顧客に価値を届けるスピードを高めるうえで、ここが構造的な制約になっていました。 問題 ：要件定義からUI設計までが少人数の設計チームに集中し、設計フェーズが価値提供スピードの制約になっていた 課題 ：UI設計を分担し、設計チームが要件の検討に集中できる状態をつくる その課題解決の案が、フロントエンドの役割の引き直しでした。 なぜフロントエンドだったのか 理由は大きく2つあると理解しています。 ひとつは、 UIを実際に動く形にできること です。フロントエンドがUI設計からプロトタイプ作成までを一気通貫で担えば、設計と検証の間の受け渡しがなくなります。 もうひとつは、 実装して初めて見えることがある という点です。実装フェーズに入り、実際に動く画面を触る中で、次のような気づきを得ることがありました。 この操作フローだと、ユーザーが迷う場面がありそうだ この情報配置だと、目的の項目にたどり着くまでに時間がかかりそうだ この入力体験は、もう少し工夫の余地がありそうだ ただ、その時点では開発プロセスもすでに後半で、操作体験を大きく変更することは難しい状態でした。 ここで起きているのは、誰かの検討が足りなかった、という話ではありません。Figmaの画面仕様は、情報設計やレイアウト、コンポーネントの選定といった判断を固めるうえで欠かせないものです。ただ、操作の連続性や入力のテンポといった「実際に動かしてみないと分からない情報」を、設計フェーズの時点で確かめる手段がありませんでした。 だとすれば、実装まで担うフロントエンドが設計段階から関わり、動く状態で確かめてしまえばいい。この2つが重なった結果としての体制変更だったと捉えています。 プロトタイプを作るコストが下がった もうひとつの後押しが、開発組織全体で進んでいる「AIを活用した開発スタイル」への転換です。 AIを活用することで、アイデアや仕様を素早く動くプロトタイプとして形にできるようになりました。これまでは「作ってから検証する」ことのコストが高く、現実的な選択肢になりにくかった。そのコストが下がったことで、設計フェーズで動かして確かめる進め方が、例外ではなく標準の選択肢になったと考えています。 プロダクトのフェーズも変わってきた 機能拡張は今も続いており、強化すべき領域は残っています。 一方で、ユーザーの声を聞く中で見えてきたのは、画面の見やすさ以上に、 日々大量の業務をどれだけストレスなく処理できるか という操作感そのものへの期待でした。 そのため、次のような観点で体験の質を磨き込むことの優先度が、以前より上がってきていると感じています。 直感性 ：初めて触るユーザーでも迷わず操作できること 効率性 ：無駄な画面遷移やステップを削ぎ落とし、大量の処理をスピーディーに行えること 信頼性 ：誤操作や確認漏れを防ぎ、日々の業務で安心して使えること 3. まずは小さくUI設計を担う 立ち上げは小さく、定着は着実に 方針が決まったとはいえ、これまで実装を中心に担ってきた私たちが、いきなり全機能・全工程を引き受けるのは現実的ではありません。 そこで ラクスリーダーシッププリンシプル（RLP） の一つである「小さく試して大きく育てる」を意識し、今後リリース予定の新機能から着手することにしました。現時点で実践したのは2案件です。 小さく始めたのはあくまで立ち上げ方の話であり、単発の試行として終わらせるつもりはありません。この2案件で得た手応えと課題をもとに、今後の新機能開発の標準にしていくことを目指しています。 Before → After 体制のBefore→After 進め方 概要設計で整理された機能要件をもとに、フロントエンドがUIを設計し、実際のコードでプロトタイプまで作り込みます。画面遷移や入力操作を本物同様に試せる状態にするのがポイントです。 設計チームが概要設計（機能要件の整理） フロントエンドがUIを設計し、プロトタイプを作成 フロントエンドチーム内でレビュー 設計チームによるレビュー・仕様の確定 UIのブラッシュアップ 本実装 UI設計そのものはフロントエンドが担いますが、 仕様の確定は設計チームとの合意を経て行います 。要件の背景や事業判断を持っているのは設計チームであり、そこと接続されていないUIは成立しないためです。役割を引き取ったというより、UI設計の検討をフロントエンド側に前倒しし、二者で詰める形に変えた、という表現が近いと思います。 実際、プロトタイプを持ち込むことで、言葉や画面仕様だけではイメージを揃えにくかった操作感について、設計チームと早い段階で具体的な議論ができるようになりました。 4. 担ってみて見えてきた、3つの課題 始めて数か月が経ちました。設計フェーズの領域に踏み込んだからこそ、向き合うことになった課題が3つあります。 課題① 顧客・業務理解を深める 最も大きな壁が、ドメイン知識と業務フローの理解でした。 実装に必要な理解と、UIを設計するために必要な理解には、思っていた以上に差がありました。「ユーザーはどういう業務の文脈で、どのタイミングでその設定を変更したくなるのか」「前後の作業とどう繋がっているのか」。ここを押さえていないと、業務に馴染むUIにはなりません。 → 営業商談の録画視聴、設計チームとのディスカッション、経理業務の専門書などを通じて、インプットを継続しています。 課題② UIパターンの引き出しを増やす 顧客の業務が理解できても、それを直感的なUIへ落とし込むには別のスキルが必要でした。 情報量の多い設定画面において、「ポップアップで出すべきか、インラインで表示すべきか」「どのように視覚的なガイドを出せば迷わないか」。こうした選定を、経験則や感覚だけで判断してしまう場面がありました。 → UI/UXデザインや各種UIパターンを学び、既存画面に積み上げられてきた判断の意図を読み解きながら、選定の引き出しを増やしています。 課題③ 設計意図を言語化する 「なぜこのUIにしたのか」を言語化し、関係者に説明する力も新たなハードルでした。 プロトタイプを持ち込んでも、「使いやすそうだから」では議論になりません。「この操作フローならユーザーの思考を妨げない」「実装コストとのバランスが良い」といった理由を、ビジネス視点も含めて説明し、合意形成を図る必要があります。 → プロトタイプを軸にした早期のすり合わせを重ね、意図を説明する力を磨いています。 3つ並べて改めて思うのは、これらはいずれも、 少人数の設計チームが日常的に引き受けてきたことの一端 だということです。自分で担ってみて初めて、その難しさを実感しました。 5. 一歩踏み出した先に見えてきた、次の景色 運用面では、詰めるべき論点も残っています。プロトタイプと本実装の境界線をどこに引くか、UI仕様のドキュメントをどう管理するか。この進め方をチームの標準として定着させるうえで、避けて通れないテーマです。 そうした中で、直近では私自身が設計メンバーとして設計チームに加わることになりました。より事業や顧客に近い場所で、課題解決や設計判断に携わっていくことになります。 実装に閉じず、顧客視点でプロダクトづくりを主導できるエンジニアになる。そこに向けた、はじめの一歩だと思っています。 おわりに 今回紹介した取り組みは、正直に言えば、まだ「成功事例」と呼べる段階ではありません。課題のほうが山積みです。それでも、「より良いプロダクトを作りたい」という思いから踏み出した以上、ここから引き返すつもりはありません。 この記事が、「もっとプロダクトの意思決定に関わりたい」と考えているフロントエンドエンジニアの方にとって、何かのヒントになれば幸いです。 設計チームの一員として見えてくる景色や、そこでの気づき・失敗についても、機会を見てまたお伝えできればと思います。