From 74229312907e87af84754bb205dfba3b5f4ce655 Mon Sep 17 00:00:00 2001 From: wuyang6 Date: Fri, 12 Jun 2026 20:17:20 +0800 Subject: [PATCH] Auto-run queued session inputs --- backend/api/server.py | 42 ++++++++++++++++++++++++++ src/session_store.py | 40 +++++++++++++++++++++++++ tests/test_session_store.py | 59 +++++++++++++++++++++++++++++++++++-- 3 files changed, 139 insertions(+), 2 deletions(-) diff --git a/backend/api/server.py b/backend/api/server.py index 16e5103..5eba318 100644 --- a/backend/api/server.py +++ b/backend/api/server.py @@ -74,6 +74,7 @@ from src.session_store import ( append_agent_display_message, agent_session_delta, cancel_session_input_queue, + consume_next_session_input, consume_session_guidance, delete_agent_session, deserialize_runtime_config, @@ -2538,6 +2539,41 @@ def create_app(state: AgentState) -> FastAPI: 'cancelled_input_queue_items': queue_cancelled, } + def _schedule_next_queued_turn( + *, + account_id: str | None, + session_id: str, + directory: Path, + ) -> dict[str, Any] | None: + item = consume_next_session_input(session_id, directory=directory) + if item is None: + return None + content = str(item.get('content') or '').strip() + if not content: + return None + try: + return _start_chat_run( + ChatRequest( + prompt=content, + account_id=account_id, + session_id=session_id, + resume_session_id=session_id, + ) + ) + except Exception as exc: # noqa: BLE001 + update_session_input_queue_item( + session_id, + int(item.get('id') or 0), + directory=directory, + status='pending', + ) + print( + f'[input-queue] failed to schedule queued turn ' + f'session={session_id} item={item.get("id")}: {exc}', + flush=True, + ) + return None + def _run_chat_payload( request: ChatRequest, event_sink: Any | None = None, @@ -2845,6 +2881,12 @@ def create_app(state: AgentState) -> FastAPI: result.session_id, status='cancelled', ) + if not run_record.cancel_event.is_set() and result.session_id: + _schedule_next_queued_turn( + account_id=request.account_id, + session_id=result.session_id, + directory=session_directory, + ) return payload def _start_chat_run(request: ChatRequest) -> dict[str, Any]: diff --git a/src/session_store.py b/src/session_store.py index fc915ae..8e427a7 100644 --- a/src/session_store.py +++ b/src/session_store.py @@ -471,6 +471,46 @@ def consume_session_guidance( return [_queue_row_to_dict(row) for row in rows] +def consume_next_session_input( + session_id: str, + directory: Path | None = None, +) -> dict[str, Any] | None: + target_dir = directory or DEFAULT_AGENT_SESSION_DIR + now = time.time() + with _connect_agent_session_db(target_dir) as conn: + row = conn.execute( + """ + select * + from agent_input_queue + where session_id = ? + and status = 'pending' + and kind = 'next_turn' + order by id asc + limit 1 + """, + (session_id,), + ).fetchone() + if row is None: + return None + cursor = conn.execute( + """ + update agent_input_queue + set status = 'consumed', + updated_at = ?, + consumed_at = ? + where id = ? and status = 'pending' + """, + (now, now, int(row['id'])), + ) + if int(cursor.rowcount or 0) <= 0: + return None + item = _queue_row_to_dict(row) + item['status'] = 'consumed' + item['updated_at'] = now + item['consumed_at'] = now + return item + + def delete_agent_session(session_id: str, directory: Path | None = None) -> bool: target_dir = directory or DEFAULT_AGENT_SESSION_DIR deleted = False diff --git a/tests/test_session_store.py b/tests/test_session_store.py index 63c2fc2..b773daf 100644 --- a/tests/test_session_store.py +++ b/tests/test_session_store.py @@ -20,8 +20,11 @@ from src.session_store import ( _deserialize_output_schema, _optional_float, _optional_int, + consume_next_session_input, deserialize_model_config, deserialize_runtime_config, + enqueue_session_input, + list_session_input_queue, load_agent_session, load_session, save_agent_session, @@ -234,6 +237,58 @@ class TestStoredAgentSessionRoundTrip(unittest.TestCase): self.assertEqual(loaded.plugin_state, {}) +class TestSessionInputQueue(unittest.TestCase): + def test_consume_next_turn_uses_fifo_and_preserves_guidance(self) -> None: + with tempfile.TemporaryDirectory() as td: + directory = Path(td) / 'sessions' + directory.mkdir() + first = enqueue_session_input( + 'session-1', + directory=directory, + content='first queued turn', + kind='next_turn', + ) + guidance = enqueue_session_input( + 'session-1', + directory=directory, + content='guide current run', + kind='guidance', + ) + second = enqueue_session_input( + 'session-1', + directory=directory, + content='second queued turn', + kind='next_turn', + ) + + consumed_first = consume_next_session_input( + 'session-1', + directory=directory, + ) + consumed_second = consume_next_session_input( + 'session-1', + directory=directory, + ) + consumed_none = consume_next_session_input( + 'session-1', + directory=directory, + ) + pending = list_session_input_queue('session-1', directory=directory) + + self.assertIsNotNone(consumed_first) + self.assertIsNotNone(consumed_second) + assert consumed_first is not None + assert consumed_second is not None + self.assertEqual(consumed_first['id'], first['id']) + self.assertEqual(consumed_first['content'], 'first queued turn') + self.assertEqual(consumed_first['status'], 'consumed') + self.assertEqual(consumed_second['id'], second['id']) + self.assertEqual(consumed_second['content'], 'second queued turn') + self.assertIsNone(consumed_none) + self.assertEqual([item['id'] for item in pending], [guidance['id']]) + self.assertEqual(pending[0]['kind'], 'guidance') + + class TestModelConfigSerialization(unittest.TestCase): """serialize_model_config + deserialize_model_config round-trip.""" @@ -389,8 +444,8 @@ class TestRuntimeConfigSerialization(unittest.TestCase): config = deserialize_runtime_config(payload) self.assertEqual(config.max_turns, 50) - self.assertAlmostEqual(config.command_timeout_seconds, 30.0) - self.assertEqual(config.max_output_chars, 12000) + self.assertAlmostEqual(config.command_timeout_seconds, 300.0) + self.assertEqual(config.max_output_chars, 50000) self.assertFalse(config.stream_model_responses) self.assertIsNone(config.auto_snip_threshold_tokens) self.assertIsNone(config.auto_compact_threshold_tokens)