diff --git a/README.md b/README.md index fd082f6e26..28fbc4a19d 100644 --- a/README.md +++ b/README.md @@ -702,6 +702,13 @@ store cannot guarantee that support. The built-in Memory and SQL stores support both fields; older custom stores with an explicit `idempotency_key` parameter continue to accept keyed input requests. +Gateway runs also assign IDs to input messages that omit one before storing the +run record and invoking the graph. This keeps the message visible as one turn +when the Web UI reconnects to an active run started by an API client. During a +rolling upgrade, route keyed runs with ID-less input and their retries to +upgraded workers until all Gateway workers are upgraded; older workers may +reject those retries with 409. See the [run API contract](backend/docs/API.md#create-run). + #### Option 2: Local Development If you prefer running services locally: diff --git a/backend/AGENTS.md b/backend/AGENTS.md index fbc27d93ee..2385b97b08 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -203,6 +203,11 @@ while `strip_internal_context_keys` scrubs both destinations. Treat `_normalize_input_messages` rejects canonical external system/developer roles; only `AUTH_SOURCE_INTERNAL` run input may retain them. +`start_run` assigns IDs to ID-less messages before recording input or invoking +the graph; keyed retries keep those IDs stable, while the pre-stamp input +remains the idempotency identity. The record and checkpoint share IDs so +frontend reconnect can deduplicate the user's message. + ## Development Workflow ### Test-Driven Development (TDD) — MANDATORY diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index e8643e8814..b84ce812bd 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -745,6 +745,40 @@ def normalize_input(raw_input: dict[str, Any] | None, *, trusted_internal: bool return result +def _assign_missing_run_input_message_ids(graph_input: dict[str, Any], *, thread_id: str, idempotency_key: str | None) -> dict[str, Any]: + """Give the run record and graph the same IDs before add_messages runs. + + Keyed retries must produce the same IDs before run admission compares their + input. Unkeyed runs get fresh IDs so repeated text remains a new turn. + Caller-supplied IDs and the original request objects are left untouched. + """ + messages = graph_input.get("messages") + if not isinstance(messages, list): + return graph_input + + used_ids = {message.id for message in messages if isinstance(message, BaseMessage) and message.id} + assigned = [] + changed = False + for index, message in enumerate(messages): + if not isinstance(message, BaseMessage) or message.id: + assigned.append(message) + continue + attempt = 0 + while True: + if idempotency_key is None: + message_id = str(uuid.uuid4()) + else: + name = f"deerflow:run-input:{thread_id}:{idempotency_key}:{index}:{attempt}" + message_id = str(uuid.uuid5(uuid.NAMESPACE_URL, name)) + if message_id not in used_ids: + break + attempt += 1 + used_ids.add(message_id) + assigned.append(message.model_copy(update={"id": message_id})) + changed = True + return {**graph_input, "messages": assigned} if changed else graph_input + + def _canonical_run_record_input( raw_input: dict[str, Any] | None, graph_input: object, @@ -2182,6 +2216,13 @@ async def start_run( graph_input = Command(resume=command["resume"]) else: graph_input = normalized_input + # Keep the idempotency identity in the caller's pre-stamp shape. A + # keyed retry can then reuse a run even though its stored run input has + # Gateway-issued IDs, and old id-less run records remain comparable. + request_input = _canonical_run_record_input(body.input, graph_input) if idempotency_key else None + knowledge_default_request_hash = hashlib.sha256(json.dumps(request_input, sort_keys=True, ensure_ascii=False).encode()).hexdigest() if idempotency_key else None + if isinstance(graph_input, dict): + graph_input = _assign_missing_run_input_message_ids(graph_input, thread_id=thread_id, idempotency_key=idempotency_key) # deerflow_trace_id is server-issued, so the caller's value is replaced # here at the trust boundary. body.metadata forks two ways -- through # build_run_config into config["metadata"], which the run worker @@ -2266,8 +2307,6 @@ async def start_run( # Keep the pre-default identity even when the agent is initially # unbound: adding a default must not reject an already-accepted retry. # The durable input still exposes the original accepted scope. - request_input = _canonical_run_record_input(body.input, graph_input) if idempotency_key else None - knowledge_default_request_hash = hashlib.sha256(json.dumps(request_input, sort_keys=True, ensure_ascii=False).encode()).hexdigest() if idempotency_key else None accepts_knowledge_default = not is_scope_recovery and not current_message_has_scope and knowledge_default_request_hash is not None admitted_knowledge_scope = admit_message_knowledge_scope( scope_graph_input, @@ -2605,7 +2644,7 @@ async def start_run( # today's default; explicit scopes and recovery do not use # this compatibility path. matches_legacy_default_request = accepts_knowledge_default and "knowledge_default_request_hash" not in stored and stored_input == request_input - matches_input = matches_default_request or matches_legacy_default_request or stored_input == body.input or stored_input == run_record_input + matches_input = matches_default_request or matches_legacy_default_request or stored_input == body.input or stored_input == request_input or stored_input == run_record_input if not matches_idempotency_request or not matches_input or record.assistant_id != body.assistant_id or stored.get("conversation_references", []) != conversation_references: raise HTTPException( status_code=409, diff --git a/backend/docs/API.md b/backend/docs/API.md index 6a35213d00..0041d10aa9 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -273,6 +273,16 @@ may be requested on a new thread after `/new`. Stateless `/api/langgraph/runs/*` endpoints do not support this header because requests without an explicit thread create a new temporary conversation. +When an `input.messages` entry has no usable `id`, the Gateway assigns one +before persisting the run input and invoking the graph. The same ID then appears +in the run record and checkpoint, so a client rejoining an active run can match +the two copies. Caller-supplied nonempty IDs are preserved. For keyed requests, +the generated IDs are stable for retries with the same thread and key; unkeyed +runs receive fresh IDs. During a rolling upgrade, route keyed requests with +ID-less input and their retries to upgraded Gateway workers until all workers +are upgraded. An older worker may reject a retry against a newly stamped run +record with 409. + **Rolling upgrades and keyed resume:** route all initial submissions and retries containing both `command.resume` and `Idempotency-Key` to upgraded Gateway workers only, across `/runs`, `/runs/stream` and `/runs/wait`, until every worker serving diff --git a/backend/tests/test_atomic_admission_worker.py b/backend/tests/test_atomic_admission_worker.py index 62647b6a44..59207bc9a0 100644 --- a/backend/tests/test_atomic_admission_worker.py +++ b/backend/tests/test_atomic_admission_worker.py @@ -32,7 +32,7 @@ def anyio_backend(): ) @pytest.mark.parametrize("with_event_store", [True, False]) async def test_real_worker_atomic_replacement_retains_finalization_authority(store, action, with_event_store, host_finalization_cancel, stale_renewal_race): - manager = RunManager(store=store, worker_id="owner", run_ownership_config=RunOwnershipConfig(heartbeat_enabled=True, lease_seconds=5, grace_seconds=0)) + manager = RunManager(store=store, worker_id="owner", run_ownership_config=RunOwnershipConfig(heartbeat_enabled=True, lease_seconds=10, grace_seconds=0)) record = await manager.create_or_reject("thread", user_id="review") started, rollback_entered, release_rollback = (asyncio.Event() for _ in range(3)) error_committed, release_error_ack = asyncio.Event(), asyncio.Event() @@ -100,9 +100,10 @@ async def test_real_worker_atomic_replacement_retains_finalization_authority(sto if stale_renewal_race: stale_renewal = asyncio.create_task(manager._renew_leases()) await asyncio.wait_for(stale_renewal_entered.wait(), 2) - # Model the final failed-renewal window before replacement using an - # actual persisted deadline; the Worker and checkpoint code stay real. - deadline = datetime.now(UTC) + timedelta(milliseconds=150) + # Model a nearing lease deadline with a real persisted timestamp. The + # replacement and rollback checkpoint may take longer than 150 ms on a + # loaded CI runner, before this test explicitly requests renewal. + deadline = datetime.now(UTC) + timedelta(seconds=4) row = await store.get(record.run_id, user_id="review") await store.put(record.run_id, thread_id="thread", status="running", user_id="review", owner_worker_id="owner", lease_expires_at=deadline.isoformat(), created_at=row["created_at"]) record.lease_expires_at = deadline.isoformat() diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index b695cfe1de..5db9d929be 100644 --- a/backend/tests/test_gateway_services.py +++ b/backend/tests/test_gateway_services.py @@ -381,6 +381,57 @@ def test_canonical_run_record_input_uses_admitted_message_snapshot(): assert stored["messages"][0]["additional_kwargs"]["knowledge_scope"]["dataset_ids"] == ["dataset-1"] +@pytest.mark.parametrize("idempotency_key", [None, "retry-key"]) +@pytest.mark.asyncio +async def test_run_input_ids_match_the_checkpoint_and_record(_stub_app_config, idempotency_key): + """An id-less API input must not become a second user turn on reconnect.""" + from unittest.mock import patch + + from langgraph.graph.message import add_messages + + from app.gateway.routers.thread_runs import RunCreateRequest + from app.gateway.services import start_run + from deerflow.runtime import RunManager + from deerflow.runtime.runs.store.memory import MemoryRunStore + + body = RunCreateRequest( + input={ + "messages": [ + {"role": "user", "content": "same question"}, + {"role": "user", "content": "same question"}, + {"role": "user", "content": "empty id", "id": ""}, + {"role": "user", "content": "client id", "id": "client-human"}, + ] + } + ) + request = _make_start_run_request(RunManager(store=MemoryRunStore())) + captured = {} + + async def fake_run_agent(*_args, **kwargs): + captured["messages"] = kwargs["graph_input"]["messages"] + + with patch("app.gateway.services.resolve_agent_factory", return_value=object()), patch("app.gateway.services.run_agent", side_effect=fake_run_agent): + record = await start_run(body, "thread-idless-input", request, idempotency_key=idempotency_key) + await record.task + + graph_messages = captured["messages"] + recorded_messages = record.kwargs["input"]["messages"] + ids = [message["id"] for message in recorded_messages] + assert len(ids) == 4 + assert all(isinstance(message_id, str) and message_id for message_id in ids) + assert len(set(ids)) == 4 + assert ids[3] == "client-human" + assert [message.id for message in graph_messages] == ids + assert [message.id for message in add_messages([], graph_messages)] == ids + assert "id" not in body.input["messages"][0] + assert body.input["messages"][2]["id"] == "" + + if idempotency_key is not None: + reused = await start_run(body, "thread-idless-input", request, idempotency_key=idempotency_key) + assert reused.run_id == record.run_id + assert [message["id"] for message in reused.kwargs["input"]["messages"]] == ids + + @pytest.mark.parametrize( "forged_original", ["spoofed audit text", [{"type": "text", "text": "spoofed audit text"}]],