diff --git a/api/oss/src/apis/fastapi/sessions/models.py b/api/oss/src/apis/fastapi/sessions/models.py index 9c4372f8e8..b6280d3896 100644 --- a/api/oss/src/apis/fastapi/sessions/models.py +++ b/api/oss/src/apis/fastapi/sessions/models.py @@ -29,7 +29,7 @@ from oss.src.core.sessions.mounts.dtos import SessionMount, SessionMountQuery from oss.src.core.sessions.turns.dtos import HarnessKind, SessionTurn, SessionTurnQuery from oss.src.core.sessions.types import SessionReference -from oss.src.core.sessions.inputs.dtos import PendingInput +from oss.src.core.sessions.inputs.dtos import PendingInputUpdate, PendingInput from oss.src.core.shared.dtos import OTelSpanId, Windowing from oss.src.dbs.postgres.sessions.streams.dao import MAX_SESSION_QUERY_LIMIT @@ -607,3 +607,7 @@ class SessionControlOutcomeResponse(BaseModel): class SessionContinuationResumeResponse(BaseModel): resumed: bool + + +class PendingInputUpdateRequest(PendingInputUpdate): + pass diff --git a/api/oss/src/apis/fastapi/sessions/router.py b/api/oss/src/apis/fastapi/sessions/router.py index b69f5231c6..96f486e81f 100644 --- a/api/oss/src/apis/fastapi/sessions/router.py +++ b/api/oss/src/apis/fastapi/sessions/router.py @@ -110,6 +110,8 @@ SessionInputIdempotencyConflict, SessionInputNotFound, SessionInputNotRemovable, + SessionInputNotEditable, + SessionInputContentInvalid, SessionInputRemoved, ) from oss.src.core.sessions.inputs.dtos import PendingInputState @@ -204,6 +206,7 @@ SessionResponse, SessionsResponse, PendingInputResponse, + PendingInputUpdateRequest, PendingInputAdmissionRequest, PendingInputAdmissionResponse, SessionCapabilities, @@ -2130,6 +2133,14 @@ def __init__( if inputs_service is not None: # The snapshot itself is `get_session_snapshot`, registered below: one route serves # both the reconnect watermark and the durable queue. + self.router.add_api_route( + "/sessions/{session_id}/inputs/{input_id}", + self.update_pending_input, + methods=["PATCH"], + operation_id="update_pending_session_input", + response_model=PendingInputResponse, + tags=["Sessions"], + ) self.router.add_api_route( "/sessions/{session_id}/inputs/{input_id}", self.remove_pending_input, @@ -2303,6 +2314,63 @@ async def query_sessions( windowing=response_windowing, ) + @intercept_exceptions() + async def update_pending_input( + self, + request: Request, + session_id: str, + input_id: UUID, + payload: PendingInputUpdateRequest, + ) -> PendingInputResponse: + _validate_session_id_http(session_id) + project_id = UUID(str(request.state.project_id)) + user_id = request.state.user_id + if not await check_action_access( + user_uid=str(user_id), + project_id=str(project_id), + permission=Permission.RUN_SESSIONS, + ): + raise FORBIDDEN_EXCEPTION + try: + item = await self.inputs_service.update( + project_id=project_id, + user_id=UUID(str(user_id)) if user_id else None, + session_id=session_id, + input_id=input_id, + update=payload, + ) + except SessionInputNotFound as error: + raise HTTPException( + status_code=404, + detail={ + "code": "pending_input_not_found", + "message": str(error), + "retryable": False, + "details": {"input_id": str(input_id)}, + }, + ) from error + except SessionInputNotEditable as error: + raise HTTPException( + status_code=409, + detail={ + "code": "pending_input_not_editable", + "message": str(error), + "retryable": False, + "details": {"input_id": str(input_id)}, + }, + ) from error + except SessionInputContentInvalid as error: + raise HTTPException( + status_code=422, + detail={ + "code": "pending_input_content_invalid", + "message": str(error), + "retryable": False, + "details": {"input_id": str(input_id)}, + }, + ) from error + return PendingInputResponse(input=item) + @intercept_exceptions() async def remove_pending_input( self, request: Request, session_id: str, input_id: UUID diff --git a/api/oss/src/core/sessions/inputs/dtos.py b/api/oss/src/core/sessions/inputs/dtos.py index bb3764a010..f86d2507d6 100644 --- a/api/oss/src/core/sessions/inputs/dtos.py +++ b/api/oss/src/core/sessions/inputs/dtos.py @@ -1,9 +1,9 @@ from datetime import datetime from enum import Enum -from typing import Any, Dict, Literal, Optional +from typing import Any, Dict, List, Literal, Optional from uuid import UUID -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict, Field from oss.src.core.shared.dtos import Identifier, Lifecycle @@ -45,3 +45,17 @@ class PendingInputPromotion(BaseModel): input: PendingInput execution_id: str created_at: datetime + + +class PendingInputAttachment(BaseModel): + model_config = ConfigDict(extra="forbid") + uri: str = Field(min_length=1) + mime_type: str = Field(min_length=1) + filename: Optional[str] = None + attachment_id: Optional[str] = Field(default=None, min_length=1) + + +class PendingInputUpdate(BaseModel): + model_config = ConfigDict(extra="forbid") + text: str + attachments: List[PendingInputAttachment] = Field(default_factory=list) diff --git a/api/oss/src/core/sessions/inputs/interfaces.py b/api/oss/src/core/sessions/inputs/interfaces.py index e4a7128d5a..575bac48ac 100644 --- a/api/oss/src/core/sessions/inputs/interfaces.py +++ b/api/oss/src/core/sessions/inputs/interfaces.py @@ -1,5 +1,5 @@ from abc import ABC, abstractmethod -from typing import Any, AsyncContextManager, List, Optional +from typing import Any, AsyncContextManager, Dict, List, Optional from uuid import UUID from oss.src.core.sessions.inputs.dtos import PendingInput, PendingInputCreate @@ -93,3 +93,22 @@ async def promote_next( transaction: Optional[Any] = None, ) -> Optional[PendingInput]: pass + + @abstractmethod + async def lock_pending_for_edit( + self, *, project_id: UUID, session_id: str, input_id: UUID, transaction: Any + ) -> Optional[PendingInput]: + pass + + @abstractmethod + async def update_content( + self, + *, + project_id: UUID, + session_id: str, + input_id: UUID, + content: Dict[str, Any], + user_id: Optional[UUID], + transaction: Any, + ) -> PendingInput: + pass diff --git a/api/oss/src/core/sessions/inputs/service.py b/api/oss/src/core/sessions/inputs/service.py index 2860c2ab77..93ff2a12a9 100644 --- a/api/oss/src/core/sessions/inputs/service.py +++ b/api/oss/src/core/sessions/inputs/service.py @@ -1,3 +1,5 @@ +from copy import deepcopy + import hashlib import json from typing import Any, Awaitable, Callable, Dict, List, Optional @@ -8,6 +10,7 @@ PendingInputAdmission, PendingInputCreate, PendingInputState, + PendingInputUpdate, ) from oss.src.core.sessions.inputs.interfaces import SessionInputsDAOInterface from oss.src.core.sessions.inputs.types import ( @@ -15,6 +18,7 @@ SessionInputIdempotencyConflict, SessionInputNotFound, SessionInputNotRemovable, + SessionInputContentInvalid, ) from oss.src.core.sessions.interactions.dtos import SessionInteractionStatus from oss.src.core.sessions.interactions.interfaces import ( @@ -35,6 +39,120 @@ def input_fingerprint(*, content: Dict[str, Any], policy: str) -> str: return hashlib.sha256(canonical).hexdigest() +def edit_pending_input_content( + content: Dict[str, Any], update: PendingInputUpdate +) -> Dict[str, Any]: + edited = deepcopy(content) + data = edited.get("data") + inputs = data.get("inputs") if isinstance(data, dict) else None + messages = inputs.get("messages") if isinstance(inputs, dict) else None + if not isinstance(messages, list): + raise SessionInputContentInvalid( + "The queued input has no editable user message." + ) + message = next( + ( + item + for item in reversed(messages) + if isinstance(item, dict) and item.get("role") == "user" + ), + None, + ) + if message is None: + raise SessionInputContentInvalid( + "The queued input has no editable user message." + ) + original = message.get("content") + field = "content" + if isinstance(original, str): + if not update.attachments: + message[field] = update.text + return edited + blocks = [{"type": "text", "text": original}] + elif isinstance(original, list): + blocks = original + elif isinstance(message.get("parts"), list): + field = "parts" + blocks = message[field] + else: + raise SessionInputContentInvalid( + "The queued user message uses an unsupported content format." + ) + kept = [] + wrote_text = False + for block in blocks: + if isinstance(block, dict) and block.get("type") == "text": + if not wrote_text: + kept.append({**block, "text": update.text}) + wrote_text = True + else: + kept.append(block) + if not wrote_text and update.text: + kept.insert(0, {"type": "text", "text": update.text}) + uris = { + block.get("uri", block.get("url")) + for block in kept + if isinstance(block, dict) + and isinstance(block.get("uri", block.get("url")), str) + } + attachment_ids = set() + for block in kept: + if not isinstance(block, dict): + continue + attachment_id = block.get("attachmentId", block.get("attachment_id")) + provider_metadata = block.get("providerMetadata") + agenta_metadata = ( + provider_metadata.get("agenta") + if isinstance(provider_metadata, dict) + else None + ) + if not attachment_id and isinstance(agenta_metadata, dict): + attachment_id = agenta_metadata.get("attachmentId") + if isinstance(attachment_id, str) and attachment_id: + attachment_ids.add(attachment_id) + for attachment in update.attachments: + if attachment.uri in uris or ( + attachment.attachment_id and attachment.attachment_id in attachment_ids + ): + continue + if field == "parts": + block = { + "type": "file", + "url": attachment.uri, + "mediaType": attachment.mime_type, + } + if attachment.attachment_id is not None: + block["providerMetadata"] = { + "agenta": {"attachmentId": attachment.attachment_id} + } + if attachment.filename is not None: + block["filename"] = attachment.filename + elif attachment.attachment_id is not None: + block = { + "type": "attachment", + "attachmentId": attachment.attachment_id, + "mimeType": attachment.mime_type, + } + if attachment.filename is not None: + block["filename"] = attachment.filename + else: + block = { + "type": "image" + if attachment.mime_type.startswith("image/") + else "resource", + "uri": attachment.uri, + "mimeType": attachment.mime_type, + } + if attachment.filename is not None: + block["filename"] = attachment.filename + kept.append(block) + uris.add(attachment.uri) + if attachment.attachment_id: + attachment_ids.add(attachment.attachment_id) + message[field] = kept + return edited + + class SessionInputsService: def __init__( self, @@ -275,3 +393,31 @@ async def remove( if existing is not None and existing.state != PendingInputState.pending: raise SessionInputNotRemovable(str(input_id)) raise SessionInputNotFound(str(input_id)) + + async def update( + self, + *, + project_id: UUID, + session_id: str, + input_id: UUID, + user_id: Optional[UUID], + update: PendingInputUpdate, + ) -> PendingInput: + async with self._dao.transaction() as transaction: + item = await self._dao.lock_pending_for_edit( + project_id=project_id, + session_id=session_id, + input_id=input_id, + transaction=transaction, + ) + if item is None: + raise SessionInputNotFound(str(input_id)) + content = edit_pending_input_content(item.content, update) + return await self._dao.update_content( + project_id=project_id, + session_id=session_id, + input_id=input_id, + content=content, + user_id=user_id, + transaction=transaction, + ) diff --git a/api/oss/src/core/sessions/inputs/types.py b/api/oss/src/core/sessions/inputs/types.py index 1f23e82b8e..0a351606e5 100644 --- a/api/oss/src/core/sessions/inputs/types.py +++ b/api/oss/src/core/sessions/inputs/types.py @@ -32,3 +32,15 @@ class SessionInputRemoved(SessionInputError): def __init__(self, input_id: str): self.input_id = input_id super().__init__("The queued input was removed and cannot be sent.") + + +class SessionInputNotEditable(SessionInputError): + def __init__(self, input_id: str): + self.input_id = input_id + super().__init__( + "The queued input is no longer editable because it was removed, promoted, or selected to run next." + ) + + +class SessionInputContentInvalid(SessionInputError): + pass diff --git a/api/oss/src/dbs/postgres/sessions/inputs/dao.py b/api/oss/src/dbs/postgres/sessions/inputs/dao.py index b31c57134f..8109657484 100644 --- a/api/oss/src/dbs/postgres/sessions/inputs/dao.py +++ b/api/oss/src/dbs/postgres/sessions/inputs/dao.py @@ -1,12 +1,15 @@ from datetime import datetime, timezone -from typing import Any, List, Optional +from typing import Any, Dict, List, Optional from uuid import UUID from sqlalchemy import and_, func, or_, select, text, update as sa_update from oss.src.core.sessions.inputs.dtos import PendingInput, PendingInputCreate from oss.src.core.sessions.inputs.interfaces import SessionInputsDAOInterface -from oss.src.core.sessions.inputs.types import SessionInputNotRemovable +from oss.src.core.sessions.inputs.types import ( + SessionInputNotRemovable, + SessionInputNotEditable, +) from oss.src.dbs.postgres.sessions.commands.dbes import SessionCommandDBE from oss.src.dbs.postgres.sessions.inputs.dbes import SessionInputDBE from oss.src.dbs.postgres.sessions.inputs.mappings import ( @@ -305,6 +308,7 @@ async def promote_next( transaction: Optional[Any] = None, ) -> Optional[PendingInput]: async def execute(session: Any) -> Optional[PendingInput]: + await self._lock_session(session, project_id, session_id) stmt = select(SessionInputDBE).where( SessionInputDBE.project_id == project_id, SessionInputDBE.session_id == session_id, @@ -333,3 +337,67 @@ async def execute(session: Any) -> Optional[PendingInput]: return await execute(transaction) async with self.engine.session() as session: return await execute(session) + + async def lock_pending_for_edit( + self, *, project_id: UUID, session_id: str, input_id: UUID, transaction: Any + ) -> Optional[PendingInput]: + await self._lock_session(transaction, project_id, session_id) + row = ( + await transaction.execute( + select(SessionInputDBE) + .where( + SessionInputDBE.project_id == project_id, + SessionInputDBE.session_id == session_id, + SessionInputDBE.id == input_id, + ) + .with_for_update() + ) + ).scalar_one_or_none() + if row is None: + return None + reserved = ( + await transaction.execute( + select(SessionCommandDBE.id) + .where( + SessionCommandDBE.project_id == project_id, + SessionCommandDBE.session_id == session_id, + SessionCommandDBE.kind == "cancel", + SessionCommandDBE.state.in_(("pending", "claimed")), + SessionCommandDBE.deleted_at.is_(None), + SessionCommandDBE.data["steer_input_id"].astext == str(input_id), + ) + .limit(1) + ) + ).scalar_one_or_none() + if row.state != "pending" or reserved is not None: + raise SessionInputNotEditable(str(input_id)) + return to_pending_input(row) + + async def update_content( + self, + *, + project_id: UUID, + session_id: str, + input_id: UUID, + content: Dict[str, Any], + user_id: Optional[UUID], + transaction: Any, + ) -> PendingInput: + row = ( + await transaction.execute( + sa_update(SessionInputDBE) + .where( + SessionInputDBE.project_id == project_id, + SessionInputDBE.session_id == session_id, + SessionInputDBE.id == input_id, + SessionInputDBE.state == "pending", + ) + .values( + content=content, + updated_at=datetime.now(timezone.utc), + updated_by_id=user_id, + ) + .returning(SessionInputDBE) + ) + ).scalar_one() + return to_pending_input(row) diff --git a/api/oss/tests/pytest/unit/sessions/test_session_inputs_dao.py b/api/oss/tests/pytest/unit/sessions/test_session_inputs_dao.py index 84f4fe7be0..5c2a8fc32f 100644 --- a/api/oss/tests/pytest/unit/sessions/test_session_inputs_dao.py +++ b/api/oss/tests/pytest/unit/sessions/test_session_inputs_dao.py @@ -24,10 +24,22 @@ from oss.src.core.sessions.commands.service import SessionCommandsService from oss.src.core.sessions.commands.types import ExecutionExpectationFailed from oss.src.core.sessions.executions.dtos import SessionExecutionState -from oss.src.core.sessions.inputs.dtos import PendingInputCreate, PendingInputState -from oss.src.core.sessions.inputs.service import SessionInputsService, input_fingerprint +from oss.src.core.sessions.inputs.dtos import ( + PendingInputCreate, + PendingInputState, + PendingInputUpdate, + PendingInputAttachment, +) +from oss.src.core.sessions.inputs.service import ( + SessionInputsService, + input_fingerprint, + edit_pending_input_content, +) from oss.src.core.sessions.inputs.types import ( SessionInputBusy, + SessionInputNotEditable, + SessionInputContentInvalid, + SessionInputNotFound, SessionInputNotRemovable, SessionInputRemoved, ) @@ -1350,3 +1362,351 @@ async def pause_reserved(**kwargs): ) removed = await inputs.remove_pending(**args) assert removed.state == PendingInputState.removed + + +@pytest.mark.asyncio +async def test_edit_pending_preserves_payload_identity_and_retry_attachments( + input_scope, +): + dao = SessionInputsDAO(engine=input_scope["engine"]) + service = SessionInputsService(inputs_dao=dao, streams_service=_BusyStreams()) + values = _input(input_scope, key="edit-existing", message="unused") + values.content = { + "data": { + "inputs": { + "messages": [ + {"role": "system", "content": "history"}, + { + "id": "user-id", + "role": "user", + "parts": [ + {"type": "text", "text": "before"}, + { + "type": "file", + "url": "agenta://old", + "mediaType": "text/plain", + "opaque": True, + }, + ], + }, + ] + }, + "parameters": {"agent": {"instructions": "keep-config"}}, + }, + "references": {"revision": {"id": "keep-revision"}}, + } + values.request_fingerprint = input_fingerprint( + content=values.content, policy=values.policy + ) + row = await dao.create_input(user_id=input_scope["user_id"], pending_input=values) + update = PendingInputUpdate( + text="after", + attachments=[ + PendingInputAttachment( + uri="agenta://new", mime_type="text/plain", filename="new.txt" + ) + ], + ) + for _ in range(2): + edited = await service.update( + project_id=input_scope["project_id"], + session_id=input_scope["session_id"], + input_id=row.id, + user_id=input_scope["user_id"], + update=update, + ) + original_retry = await service.admit( + project_id=input_scope["project_id"], + session_id=input_scope["session_id"], + user_id=input_scope["user_id"], + content=values.content, + policy=values.policy, + idempotency_key=values.idempotency_key, + ) + assert original_retry.input.id == row.id + assert original_retry.input.content == edited.content + assert edited.id == row.id and edited.position == row.position + assert ( + edited.request_fingerprint == row.request_fingerprint + and edited.idempotency_key == row.idempotency_key + ) + assert edited.content["references"] == values.content["references"] + assert edited.content["data"]["parameters"] == values.content["data"]["parameters"] + messages = edited.content["data"]["inputs"]["messages"] + assert messages[0] == values.content["data"]["inputs"]["messages"][0] + assert messages[1]["id"] == "user-id" + assert messages[1]["parts"] == [ + {"type": "text", "text": "after"}, + { + "type": "file", + "url": "agenta://old", + "mediaType": "text/plain", + "opaque": True, + }, + { + "type": "file", + "url": "agenta://new", + "mediaType": "text/plain", + "filename": "new.txt", + }, + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("row_state", ["pending", "claimed", "promoted", "removed"]) +async def test_edit_pending_rejects_promoted_and_reserved_rows(input_scope, row_state): + dao = SessionInputsDAO(engine=input_scope["engine"]) + service = SessionInputsService(inputs_dao=dao, streams_service=_BusyStreams()) + row = await dao.create_input( + user_id=input_scope["user_id"], + pending_input=_input(input_scope, key="edit-reserved", message="original"), + ) + if row_state in ("pending", "claimed"): + await _pending_command(input_scope, data={"steer_input_id": str(row.id)}) + if row_state == "claimed": + async with input_scope["engine"].session() as tx: + await tx.execute( + text( + "UPDATE session_commands SET state='claimed' WHERE project_id=:project" + ), + {"project": input_scope["project_id"]}, + ) + else: + async with input_scope["engine"].session() as tx: + await tx.execute( + text("UPDATE session_inputs SET state=:state WHERE id=:id"), + {"state": row_state, "id": row.id}, + ) + with pytest.raises(SessionInputNotEditable): + await service.update( + project_id=input_scope["project_id"], + session_id=input_scope["session_id"], + input_id=row.id, + user_id=input_scope["user_id"], + update=PendingInputUpdate(text="changed"), + ) + stored = await dao.fetch_input( + project_id=input_scope["project_id"], + session_id=input_scope["session_id"], + input_id=row.id, + ) + assert stored.content == row.content + + +@pytest.mark.asyncio +async def test_promotion_waits_for_edited_head_instead_of_skipping_it(input_scope): + dao = SessionInputsDAO(engine=input_scope["engine"]) + first = await dao.create_input( + user_id=input_scope["user_id"], + pending_input=_input(input_scope, key="edit-first", message="first"), + ) + await dao.create_input( + user_id=input_scope["user_id"], + pending_input=_input(input_scope, key="edit-second", message="second"), + ) + async with dao.transaction() as tx: + await dao.lock_pending_for_edit( + project_id=input_scope["project_id"], + session_id=input_scope["session_id"], + input_id=first.id, + transaction=tx, + ) + promotion = asyncio.create_task( + dao.promote_next( + project_id=input_scope["project_id"], + session_id=input_scope["session_id"], + execution_id="next", + ) + ) + await asyncio.sleep(0.05) + assert not promotion.done() + await dao.update_content( + project_id=input_scope["project_id"], + session_id=input_scope["session_id"], + input_id=first.id, + content={"edited": "head"}, + user_id=input_scope["user_id"], + transaction=tx, + ) + promoted = await asyncio.wait_for(promotion, 2) + assert promoted.id == first.id + assert promoted.content == {"edited": "head"} + + +@pytest.mark.asyncio +async def test_edit_pending_scope_and_invalid_content_leave_row_unchanged(input_scope): + dao = SessionInputsDAO(engine=input_scope["engine"]) + service = SessionInputsService(inputs_dao=dao, streams_service=_BusyStreams()) + row = await dao.create_input( + user_id=input_scope["user_id"], + pending_input=_input(input_scope, key="invalid-edit", message="opaque"), + ) + args = dict( + project_id=input_scope["project_id"], + session_id=input_scope["session_id"], + input_id=row.id, + user_id=input_scope["user_id"], + update=PendingInputUpdate(text="new"), + ) + with pytest.raises(SessionInputNotFound): + await service.update(**{**args, "project_id": uuid.uuid4()}) + with pytest.raises(SessionInputContentInvalid): + await service.update(**args) + stored = await dao.fetch_input( + project_id=args["project_id"], session_id=args["session_id"], input_id=row.id + ) + assert stored.content == row.content + + +@pytest.mark.parametrize( + "original", + [ + "before", + [ + {"type": "text", "text": "before"}, + {"type": "attachment", "uri": "agenta://old", "opaque": True}, + ], + ], +) +def test_edit_pending_canonical_content_keeps_attachments(original): + with pytest.raises(ValueError): + PendingInputAttachment( + uri="agenta://invalid", + mime_type="text/plain", + attachment_id="", + ) + content = { + "data": {"inputs": {"messages": [{"role": "user", "content": original}]}} + } + update = PendingInputUpdate( + text="after", + attachments=[ + PendingInputAttachment(uri="agenta://new", mime_type="text/plain") + ], + ) + edited = edit_pending_input_content(content, update) + assert edit_pending_input_content(edited, update) == edited + blocks = edited["data"]["inputs"]["messages"][0]["content"] + assert blocks[0] == {"type": "text", "text": "after"} + assert blocks[-1] == { + "type": "resource", + "uri": "agenta://new", + "mimeType": "text/plain", + } + if isinstance(original, list): + assert blocks[1] == original[1] + assert content["data"]["inputs"]["messages"][0]["content"] == original + + +def test_edit_pending_canonical_content_preserves_durable_attachment_identity(): + old_attachment_id = "01995d1a-2f83-7c4d-8a6b-123456789abc" + new_attachment_id = "01996b6c-7b6b-7000-8000-000000000001" + original_attachment = { + "type": "attachment", + "attachmentId": old_attachment_id, + "mimeType": "application/pdf", + "filename": "old.pdf", + } + content = { + "data": { + "inputs": { + "messages": [ + { + "role": "user", + "content": [ + {"type": "text", "text": "before"}, + original_attachment, + ], + } + ] + } + } + } + update = PendingInputUpdate( + text="after", + attachments=[ + PendingInputAttachment( + uri="https://files.test/old.pdf", + mime_type="application/pdf", + filename="old.pdf", + attachment_id=old_attachment_id, + ), + PendingInputAttachment( + uri="https://files.test/new.png", + mime_type="image/png", + filename="new.png", + attachment_id=new_attachment_id, + ), + ], + ) + + edited = edit_pending_input_content(content, update) + assert edit_pending_input_content(edited, update) == edited + assert edited["data"]["inputs"]["messages"][0]["content"] == [ + {"type": "text", "text": "after"}, + original_attachment, + { + "type": "attachment", + "attachmentId": new_attachment_id, + "mimeType": "image/png", + "filename": "new.png", + }, + ] + + +def test_edit_pending_ui_parts_preserves_durable_attachment_identity(): + old_attachment_id = "01995d1a-2f83-7c4d-8a6b-123456789abc" + new_attachment_id = "01996b6c-7b6b-7000-8000-000000000001" + original_file = { + "type": "file", + "url": "https://files.test/old.pdf", + "mediaType": "application/pdf", + "filename": "old.pdf", + "providerMetadata": {"agenta": {"attachmentId": old_attachment_id}}, + } + content = { + "data": { + "inputs": { + "messages": [ + { + "role": "user", + "parts": [ + {"type": "text", "text": "before"}, + original_file, + ], + } + ] + } + } + } + update = PendingInputUpdate( + text="after", + attachments=[ + PendingInputAttachment( + uri="https://other-host.test/old.pdf", + mime_type="application/pdf", + filename="old.pdf", + attachment_id=old_attachment_id, + ), + PendingInputAttachment( + uri="https://files.test/new.png", + mime_type="image/png", + filename="new.png", + attachment_id=new_attachment_id, + ), + ], + ) + + edited = edit_pending_input_content(content, update) + assert edit_pending_input_content(edited, update) == edited + assert edited["data"]["inputs"]["messages"][0]["parts"] == [ + {"type": "text", "text": "after"}, + original_file, + { + "type": "file", + "url": "https://files.test/new.png", + "mediaType": "image/png", + "filename": "new.png", + "providerMetadata": {"agenta": {"attachmentId": new_attachment_id}}, + }, + ] diff --git a/web/mobile/src/features/chat/LiveConversation.tsx b/web/mobile/src/features/chat/LiveConversation.tsx index 077431f55b..0c0359d1cb 100644 --- a/web/mobile/src/features/chat/LiveConversation.tsx +++ b/web/mobile/src/features/chat/LiveConversation.tsx @@ -621,7 +621,7 @@ export const LiveConversation = ({ /> {/* What you have lined up stays visible while a gate is open: the queued message is the acknowledgement that the user's Send was not lost. */} - {conversation.queued.length > 0 ? ( + {conversation.queued.length > 0 || conversation.editingId ? (
{ - const open = queued.length > 0 + const open = queued.length > 0 || !!editingId // Latch the last non-empty queue: emptying it starts the collapse, and without this the rows // would vanish first and leave an empty box folding shut. const shownRef = useRef(queued) diff --git a/web/packages/agenta-api-client/src/generated/api/resources/sessions/client/Client.ts b/web/packages/agenta-api-client/src/generated/api/resources/sessions/client/Client.ts index 1cce2ad179..67a332dabf 100644 --- a/web/packages/agenta-api-client/src/generated/api/resources/sessions/client/Client.ts +++ b/web/packages/agenta-api-client/src/generated/api/resources/sessions/client/Client.ts @@ -2803,6 +2803,86 @@ export class SessionsClient { return handleNonStatusCodeError(_response.error, _response.rawResponse, "GET", "/sessions/{session_id}"); } + /** + * @param {AgentaApi.PendingInputUpdateRequest} request + * @param {SessionsClient.RequestOptions} requestOptions - Request-specific configuration. + * + * @throws {@link AgentaApi.UnprocessableEntityError} + * + * @example + * await client.sessions.updatePendingSessionInput({ + * session_id: "session_id", + * input_id: "input_id", + * text: "text" + * }) + */ + public updatePendingSessionInput( + request: AgentaApi.PendingInputUpdateRequest, + requestOptions?: SessionsClient.RequestOptions, + ): core.HttpResponsePromise { + return core.HttpResponsePromise.fromPromise(this.__updatePendingSessionInput(request, requestOptions)); + } + + private async __updatePendingSessionInput( + request: AgentaApi.PendingInputUpdateRequest, + requestOptions?: SessionsClient.RequestOptions, + ): Promise> { + const { session_id: sessionId, input_id: inputId, ..._body } = request; + const _authRequest: core.AuthRequest = await this._options.authProvider.getAuthRequest(); + const _headers: core.Fetcher.Args["headers"] = mergeHeaders( + _authRequest.headers, + this._options?.headers, + requestOptions?.headers, + ); + const _response = await core.fetcher({ + url: core.url.join( + (await core.Supplier.get(this._options.baseUrl)) ?? + (await core.Supplier.get(this._options.environment)) ?? + environments.AgentaApiEnvironment.Default, + `sessions/${core.url.encodePathParam(sessionId)}/inputs/${core.url.encodePathParam(inputId)}`, + ), + method: "PATCH", + headers: _headers, + contentType: "application/json", + queryParameters: requestOptions?.queryParams, + requestType: "json", + body: _body, + timeoutMs: (requestOptions?.timeoutInSeconds ?? this._options?.timeoutInSeconds ?? 30) * 1000, + maxRetries: requestOptions?.maxRetries ?? this._options?.maxRetries, + withCredentials: true, + abortSignal: requestOptions?.abortSignal, + fetchFn: this._options?.fetch, + logging: this._options.logging, + }); + if (_response.ok) { + return { data: _response.body as AgentaApi.PendingInputResponse, rawResponse: _response.rawResponse }; + } + + if (_response.error.reason === "status-code") { + switch (_response.error.statusCode) { + case 422: + throw new AgentaApi.UnprocessableEntityError( + _response.error.body as AgentaApi.HttpValidationError, + _response.rawResponse, + ); + default: + throw new errors.AgentaApiError({ + statusCode: _response.error.statusCode, + body: _response.error.body, + rawResponse: _response.rawResponse, + }); + } + } + + return handleNonStatusCodeError( + _response.error, + _response.rawResponse, + "PATCH", + "/sessions/{session_id}/inputs/{input_id}", + ); + } + + /** * @param {AgentaApi.SendPendingSessionInputNowRequest} request * @param {SessionsClient.RequestOptions} requestOptions - Request-specific configuration. diff --git a/web/packages/agenta-api-client/src/generated/api/resources/sessions/client/requests/PendingInputUpdateRequest.ts b/web/packages/agenta-api-client/src/generated/api/resources/sessions/client/requests/PendingInputUpdateRequest.ts new file mode 100644 index 0000000000..f02ff509a9 --- /dev/null +++ b/web/packages/agenta-api-client/src/generated/api/resources/sessions/client/requests/PendingInputUpdateRequest.ts @@ -0,0 +1,18 @@ +// This file was auto-generated by Fern from our API Definition. + +import type * as AgentaApi from "../../../../index.js"; + +/** + * @example + * { + * session_id: "session_id", + * input_id: "input_id", + * text: "text" + * } + */ +export interface PendingInputUpdateRequest { + session_id: string; + input_id: string; + text: string; + attachments?: AgentaApi.PendingInputAttachment[]; +} diff --git a/web/packages/agenta-api-client/src/generated/api/resources/sessions/client/requests/index.ts b/web/packages/agenta-api-client/src/generated/api/resources/sessions/client/requests/index.ts index 3baf4850e7..b1a4996636 100644 --- a/web/packages/agenta-api-client/src/generated/api/resources/sessions/client/requests/index.ts +++ b/web/packages/agenta-api-client/src/generated/api/resources/sessions/client/requests/index.ts @@ -37,3 +37,4 @@ export type { UnarchiveSessionRequest } from "./UnarchiveSessionRequest.js"; export type { WatchProjectRequest } from "./WatchProjectRequest.js"; export type { WatchSessionStreamRequest } from "./WatchSessionStreamRequest.js"; export type { SendPendingSessionInputNowRequest } from "./SendPendingSessionInputNowRequest.js"; +export { type PendingInputUpdateRequest } from "./PendingInputUpdateRequest.js"; diff --git a/web/packages/agenta-api-client/src/generated/api/types/PendingInputAttachment.ts b/web/packages/agenta-api-client/src/generated/api/types/PendingInputAttachment.ts new file mode 100644 index 0000000000..de6a34b0ac --- /dev/null +++ b/web/packages/agenta-api-client/src/generated/api/types/PendingInputAttachment.ts @@ -0,0 +1,8 @@ +// This file was auto-generated by Fern from our API Definition. + +export interface PendingInputAttachment { + uri: string; + mime_type: string; + filename?: (string | null) | undefined; + attachment_id?: (string | null) | undefined; +} diff --git a/web/packages/agenta-api-client/src/generated/api/types/index.ts b/web/packages/agenta-api-client/src/generated/api/types/index.ts index 9eca31cdcf..21811f862d 100644 --- a/web/packages/agenta-api-client/src/generated/api/types/index.ts +++ b/web/packages/agenta-api-client/src/generated/api/types/index.ts @@ -718,3 +718,4 @@ export * from "./WorkspaceMemberResponse.js"; export * from "./WorkspacePermission.js"; export * from "./WorkspaceResponse.js"; export * from "./PendingInputAdmissionResponse.js"; +export * from "./PendingInputAttachment.js"; diff --git a/web/packages/agenta-chat/src/assets/pendingInputs.ts b/web/packages/agenta-chat/src/assets/pendingInputs.ts index bd6dc62936..f585c5f1f2 100644 --- a/web/packages/agenta-chat/src/assets/pendingInputs.ts +++ b/web/packages/agenta-chat/src/assets/pendingInputs.ts @@ -3,6 +3,8 @@ import type {FileUIPart} from "ai" import type {QueuedMessage} from "../hooks/useAgentChatQueue" +import {attachmentContentUrl} from "./transcriptToMessages" + export interface SessionPendingInputView { capabilities: {queue: boolean; steer: boolean} executionState: "idle" | "running" | "stopping" @@ -14,19 +16,33 @@ const asRecord = (value: unknown): Record | null => ? (value as Record) : null -const filePartFromBlock = (block: Record): FileUIPart | null => { - const url = block.uri ?? block.url +const filePartFromBlock = ( + block: Record, + sessionId: string, +): FileUIPart | null => { + const metadata = asRecord(block.providerMetadata) + const agenta = asRecord(metadata?.agenta) + const attachmentId = block.attachmentId ?? block.attachment_id ?? agenta?.attachmentId + const reference = typeof attachmentId === "string" && attachmentId ? attachmentId : null + const url = reference ? attachmentContentUrl(sessionId, reference) : (block.uri ?? block.url) if (typeof url !== "string" || !url) return null + const mediaType = block.mimeType ?? block.mime_type ?? block.mediaType + const size = block.size ?? agenta?.size return { type: "file", url, - mediaType: - typeof block.mime_type === "string" - ? block.mime_type - : typeof block.mediaType === "string" - ? block.mediaType - : "application/octet-stream", + mediaType: typeof mediaType === "string" ? mediaType : "application/octet-stream", filename: typeof block.filename === "string" ? block.filename : undefined, + ...(reference + ? { + providerMetadata: { + agenta: { + attachmentId: reference, + ...(typeof size === "number" ? {size} : {}), + }, + }, + } + : {}), } } @@ -52,7 +68,7 @@ export const pendingInputToQueuedMessage = (input: PendingSessionInput): QueuedM if (block.type === "text" && typeof block.text === "string") text += block.text if (["attachment", "image", "resource"].includes(String(block.type))) { attachmentCount += 1 - const part = filePartFromBlock(block) + const part = filePartFromBlock(block, input.session_id) if (part) fileParts.push(part) } } @@ -63,7 +79,7 @@ export const pendingInputToQueuedMessage = (input: PendingSessionInput): QueuedM if (part.type === "text" && typeof part.text === "string") text += part.text if (part.type === "file") { attachmentCount += 1 - const filePart = filePartFromBlock(part) + const filePart = filePartFromBlock(part, input.session_id) if (filePart) fileParts.push(filePart) } } @@ -76,7 +92,7 @@ export const pendingInputToQueuedMessage = (input: PendingSessionInput): QueuedM attachmentCount, policy: input.policy, source: "server", - editable: false, + editable: input.state === "pending", } } diff --git a/web/packages/agenta-chat/src/components/QueuedMessagesDock.tsx b/web/packages/agenta-chat/src/components/QueuedMessagesDock.tsx index 6b23cdcd7b..7f85477bda 100644 --- a/web/packages/agenta-chat/src/components/QueuedMessagesDock.tsx +++ b/web/packages/agenta-chat/src/components/QueuedMessagesDock.tsx @@ -125,16 +125,6 @@ const Row = ({ {attachmentCount ? "(attachments only)" : "(empty message)"} )} - {attachmentCount > files.length ? ( - - {attachmentCount} attachment{attachmentCount === 1 ? "" : "s"} - - ) : null} - {message.policy === "steer" ? ( - - Steer - - ) : null} {/* Keep Send Now visible without hover. */} - Send Now + {sending || message.policy === "steer" ? "Sending" : "Send Now"} ) : null} {editing ? ( @@ -251,6 +241,8 @@ const QueuedMessagesDock = ({ ? "relative after:absolute after:-inset-x-1 after:-inset-y-2 after:content-['']" : "" + const editingMissingRow = !!editingId && !queued.some((message) => message.id === editingId) + return (
{/* px-3 so the icon starts on the same 13px line as the row text below it and the @@ -279,6 +271,14 @@ const QueuedMessagesDock = ({ />
+ {editingMissingRow ? ( +
+ This message is no longer queued. + +
+ ) : null} {/* The composer sits directly below, so a hard mount/unmount teleports it by the body's full height. `HeightCollapse` is the app's one collapse primitive — the same motion as the accordion sections and the sibling docks — and it owns aria-hidden diff --git a/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts b/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts index 4e147ffed9..733a5dc688 100644 --- a/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts +++ b/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts @@ -33,6 +33,7 @@ export interface ServerQueueAdapter { submit: (message: QueuedMessage, policy: "queue" | "steer") => Promise remove: (id: string) => Promise sendNow?: (id: string) => Promise + edit?: (id: string, item: {text: string; fileParts?: FileUIPart[]}) => Promise } interface UseAgentChatQueueArgs { @@ -214,9 +215,19 @@ export const useAgentChatQueue = ({ return message }, []) + const [editingId, setEditingId] = useState(null) + const stashRef = useRef("") + const editSessionRef = useRef<{id: string; server: boolean} | null>(null) + // A message held before an approval answer predates the server-owned continuation. Move it // under the same durable admission before that continuation can promote a different input. const migrationRef = useRef(null) + const migrationPromiseRef = useRef<{ + id: string + promise: Promise + retry: () => Promise + failed: boolean + } | null>(null) const migrationRetryTimerRef = useRef | null>(null) const [migrationRetry, setMigrationRetry] = useState(0) useEffect( @@ -234,14 +245,16 @@ export const useAgentChatQueue = ({ !server?.capabilities.queue || !submitToServer || !head || + editingId === head.id || migrationRef.current ) { return } migrationRef.current = head.id - void submitToServer(head, "queue") - .then(() => { + const retry = () => + submitToServer(head, "queue").then(() => { + if (editSessionRef.current?.id === head.id) editSessionRef.current.server = true if (sessionId) { const stored = queuedBySession.get(sessionId) if (stored) { @@ -252,7 +265,11 @@ export const useAgentChatQueue = ({ } setQueued((items) => items.filter((item) => item.id !== head.id)) }) + const migration = {id: head.id, promise: retry(), retry, failed: false} + migrationPromiseRef.current = migration + void migration.promise .catch(() => { + migration.failed = true if (migrationRef.current !== head.id) return migrationRef.current = null migrationRetryTimerRef.current = setTimeout(() => { @@ -263,10 +280,13 @@ export const useAgentChatQueue = ({ }) .finally(() => { if (migrationRef.current === head.id) migrationRef.current = null + if (migrationPromiseRef.current === migration && !migration.failed) + migrationPromiseRef.current = null }) }, [ continuationExecutionId, continuationHold, + editingId, migrationRetry, queued, sessionId, @@ -343,14 +363,19 @@ export const useAgentChatQueue = ({ // An edit session BORROWS the composer: the target's text goes in, and whatever the user had // already typed is stashed and handed back when the session ends (either way). Without that, // clicking edit on a half-written message would silently destroy it. - const [editingId, setEditingId] = useState(null) - const stashRef = useRef("") /** Open a session on `id`, stashing the composer's current draft. */ - const beginEdit = useCallback((id: string, draft = "") => { - stashRef.current = draft - setEditingId(id) - }, []) + const beginEdit = useCallback( + (id: string, draft = "") => { + editSessionRef.current = { + id, + server: !!server?.queued.some((message) => message.id === id), + } + stashRef.current = draft + setEditingId(id) + }, + [server], + ) /** Take the stashed draft back, once. Both ends of a session hand the composer back. */ const takeStash = useCallback(() => { @@ -361,6 +386,7 @@ export const useAgentChatQueue = ({ /** Close the session without touching the message. Returns the draft to restore. */ const cancelEdit = useCallback(() => { + editSessionRef.current = null setEditingId(null) return takeStash() }, [takeStash]) @@ -373,8 +399,7 @@ export const useAgentChatQueue = ({ * Attachments MERGE rather than replace — the composer only submits newly staged files, so * replacing would delete the queued message's originals on every text-only edit. * - * The queue drains on its own, so the target can leave mid-edit. Nothing is left to rewrite - * then, and the content becomes a new queued message instead of vanishing. + * A drained local target becomes a new message; durable edits instead preserve server refusal. * * Returns the stashed draft, exactly as `cancelEdit` does: committing consumes the composer, * so the text the session displaced has to come back here too or it is lost for good. @@ -382,6 +407,45 @@ export const useAgentChatQueue = ({ const commitEdit = useCallback( (item: {text: string; fileParts?: FileUIPart[]; stagedFiles?: ComposerAttachment[]}) => { const id = editingId + const editSession = editSessionRef.current + const serverOwnsInput = + editSession?.server || server?.queued.some((message) => message.id === id) + if (id && serverOwnsInput) { + if (editSession) editSession.server = true + setQueued((queue) => queue.filter((message) => message.id !== id)) + if (migrationRef.current === id) migrationRef.current = null + if (migrationPromiseRef.current?.id === id) migrationPromiseRef.current = null + } + const migration = + migrationPromiseRef.current?.id === id ? migrationPromiseRef.current : null + if (id && (serverOwnsInput || migration)) { + const save = server?.edit + if (!save) return Promise.reject(new Error("This queued message cannot be edited.")) + if (migration?.failed) { + migration.failed = false + migration.promise = migration.retry().catch((error: unknown) => { + migration.failed = true + throw error + }) + } + const saved = migration + ? migration.promise.then(() => + editSessionRef.current === editSession ? save(id, item) : undefined, + ) + : save(id, item) + return saved.then( + () => { + if (editSessionRef.current !== editSession) return "" + editSessionRef.current = null + setEditingId(null) + return takeStash() + }, + (error: unknown) => { + if (editSessionRef.current !== editSession) return "" + throw error + }, + ) + } const target = id ? queuedRef.current.find((m) => m.id === id) : undefined if (!target) { const submission = submit(item) @@ -414,7 +478,7 @@ export const useAgentChatQueue = ({ ) return draft }, - [editingId, submit, takeStash], + [editingId, server, submit, takeStash], ) // Release the queue head once the stream settles; the latch caps it at one per settle. Both diff --git a/web/packages/agenta-chat/src/hooks/useServerSessionInputs.ts b/web/packages/agenta-chat/src/hooks/useServerSessionInputs.ts index 039c78413c..373367e350 100644 --- a/web/packages/agenta-chat/src/hooks/useServerSessionInputs.ts +++ b/web/packages/agenta-chat/src/hooks/useServerSessionInputs.ts @@ -5,12 +5,14 @@ import { fetchSessionSnapshotAtom, removePendingSessionInputAtom, sendPendingSessionInputNowAtom, + updatePendingSessionInputAtom, } from "@agenta/entities/session" import {buildAgentRequest} from "@agenta/playground/agent-chat" import {projectIdAtom} from "@agenta/shared/state" -import type {UIMessage} from "ai" +import type {FileUIPart, UIMessage} from "ai" import {useAtomValue, useSetAtom} from "jotai" +import {attachmentIdForPart} from "../assets/files" import {reduceSessionPendingInputs, type SessionPendingInputView} from "../assets/pendingInputs" import type {QueuedMessage} from "./useAgentChatQueue" @@ -23,6 +25,7 @@ export interface ServerSessionInputs { submit: (message: QueuedMessage, policy: "queue" | "steer") => Promise remove: (id: string) => Promise sendNow: (id: string) => Promise + edit: (id: string, item: {text: string; fileParts?: FileUIPart[]}) => Promise refresh: () => Promise resolveCapabilities: () => Promise } @@ -53,6 +56,7 @@ export const useServerSessionInputs = ({ const fetchCapabilities = useSetAtom(fetchSessionCapabilitiesAtom) const removeInput = useSetAtom(removePendingSessionInputAtom) const sendInputNow = useSetAtom(sendPendingSessionInputNowAtom) + const updateInput = useSetAtom(updatePendingSessionInputAtom) const [viewState, setViewState] = useState<{scope: string; view: SessionPendingInputView}>( () => ({scope, view: emptyView}), ) @@ -187,6 +191,26 @@ export const useServerSessionInputs = ({ [refresh, removeInput, sessionId], ) + const edit = useCallback( + async (id: string, item: {text: string; fileParts?: FileUIPart[]}) => { + if (!view.capabilities.queue) throw new Error("Queue editing is not available.") + const updated = await updateInput({ + sessionId, + inputId: id, + text: item.text, + attachments: item.fileParts?.map((part) => ({ + uri: part.url, + mime_type: part.mediaType, + attachment_id: attachmentIdForPart(part) ?? undefined, + ...(part.filename ? {filename: part.filename} : {}), + })), + }) + if (!updated) throw new Error("The queued message could not be updated. Try again.") + await refresh() + }, + [refresh, sessionId, updateInput, view.capabilities.queue], + ) + const sendNow = useCallback( async (id: string) => { if (!view.capabilities.queue || !view.capabilities.steer) { @@ -208,6 +232,7 @@ export const useServerSessionInputs = ({ submit, remove, sendNow, + edit, refresh, resolveCapabilities, } diff --git a/web/packages/agenta-chat/tests/unit/QueuedMessagesDock.test.tsx b/web/packages/agenta-chat/tests/unit/QueuedMessagesDock.test.tsx index 3c8e463b9f..3ad9a88d93 100644 --- a/web/packages/agenta-chat/tests/unit/QueuedMessagesDock.test.tsx +++ b/web/packages/agenta-chat/tests/unit/QueuedMessagesDock.test.tsx @@ -1,5 +1,9 @@ +// @vitest-environment jsdom import {renderToStaticMarkup} from "react-dom/server" -import {describe, expect, it} from "vitest" +import {cleanup, fireEvent, render, screen} from "@testing-library/react" +import {afterEach, describe, expect, it, vi} from "vitest" + +afterEach(cleanup) import QueuedMessagesDock from "../../src/components/QueuedMessagesDock" @@ -17,3 +21,20 @@ describe("QueuedMessagesDock", () => { expect(markup).toContain("continue afterward") }) }) + +it.each([false, true])( + "keeps cancel editing reachable after the edited row leaves (touch=%s)", + (touch) => { + const cancel = vi.fn() + const props = {onRemove: vi.fn(), onCancelEdit: cancel, editingId: "edited", touch} + const {rerender} = render( + , + ) + fireEvent.click(screen.getByRole("button", {name: "Collapse"})) + rerender() + expect(screen.getByText("This message is no longer queued.")).toBeTruthy() + fireEvent.click(screen.getByRole("button", {name: "Cancel editing"})) + expect(cancel).toHaveBeenCalledOnce() + expect(props.onRemove).not.toHaveBeenCalled() + }, +) diff --git a/web/packages/agenta-chat/tests/unit/assets/pendingInputs.test.ts b/web/packages/agenta-chat/tests/unit/assets/pendingInputs.test.ts index f6a77f869f..34ec2dea64 100644 --- a/web/packages/agenta-chat/tests/unit/assets/pendingInputs.test.ts +++ b/web/packages/agenta-chat/tests/unit/assets/pendingInputs.test.ts @@ -48,7 +48,7 @@ describe("pending input reducer", () => { ]) }) - it("reduces neutral text and attachment blocks without making server rows editable", () => { + it("makes pending rows editable and retains uploaded attachment references", () => { const queued = pendingInputToQueuedMessage( input("input-1", 1, [ {type: "text", text: "Check this"}, @@ -62,9 +62,18 @@ describe("pending input reducer", () => { text: "Check this", attachmentCount: 2, source: "server", - editable: false, + editable: true, }) expect(queued?.fileParts).toEqual([ + { + type: "file", + url: expect.stringContaining( + "/sessions/attachments/asset-1/content?session_id=session-1", + ), + mediaType: "application/octet-stream", + filename: "brief.pdf", + providerMetadata: {agenta: {attachmentId: "asset-1"}}, + }, { type: "file", url: "https://files.test/image.png", @@ -74,6 +83,42 @@ describe("pending input reducer", () => { ]) }) + it.each(["content", "parts"])("preserves durable file identity from %s", (field) => { + const attachmentId = "01995d1a-2f83-7c4d-8a6b-123456789abc" + const row = input("input-1", 1, "") + const block = + field === "content" + ? { + type: "attachment", + attachmentId, + mimeType: "text/plain", + filename: "notes.txt", + size: 42, + } + : { + type: "file", + url: "https://old-host.test/content", + mediaType: "text/plain", + filename: "notes.txt", + providerMetadata: {agenta: {attachmentId, size: 42}}, + } + row.content.data.inputs.messages = [ + {role: "user", [field]: [block]}, + ] as typeof row.content.data.inputs.messages + const queued = pendingInputToQueuedMessage(row) + expect(queued?.fileParts).toEqual([ + { + type: "file", + url: expect.stringContaining( + `/sessions/attachments/${attachmentId}/content?session_id=session-1`, + ), + mediaType: "text/plain", + filename: "notes.txt", + providerMetadata: {agenta: {attachmentId, size: 42}}, + }, + ]) + }) + it("keeps a promoted input visible while its continuation is recoverable", () => { const recoverable = input("input-1", 1, "retry me", "queue", "promoted") @@ -91,7 +136,12 @@ describe("pending input reducer", () => { }) expect(view.queued).toEqual([ - expect.objectContaining({id: "input-1", text: "retry me", source: "server"}), + expect.objectContaining({ + id: "input-1", + text: "retry me", + source: "server", + editable: false, + }), ]) }) diff --git a/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts b/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts index 264daf942e..d4f0d8ae68 100644 --- a/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts +++ b/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts @@ -785,6 +785,114 @@ describe("useAgentChatQueue: reclaiming a sent message", () => { }) }) +describe("durable queued edits", () => { + it("keeps the edit and draft until same-row persistence succeeds, including a retry", async () => { + const edit = vi + .fn() + .mockRejectedValueOnce(new Error("conflict")) + .mockResolvedValueOnce(undefined) + const server: ServerQueueAdapter = { + capabilities: {queue: true, steer: true}, + busy: true, + queued: [ + {id: "first", text: "first", source: "server"}, + {id: "selected", text: "old", source: "server"}, + ], + submit: vi.fn(), + remove: vi.fn(), + edit, + } + const {result, sendQueued} = setup({...settledEmpty, server}) + act(() => result.current.beginEdit("selected", "original draft")) + await act(async () => { + await expect(result.current.commitEdit({text: "new"})).rejects.toThrow("conflict") + }) + expect(result.current.editingId).toBe("selected") + expect(result.current.queued.map((row) => row.id)).toEqual(["first", "selected"]) + let restored: string | undefined + await act(async () => { + restored = await result.current.commitEdit({text: "new"}) + }) + expect(restored).toBe("original draft") + expect(result.current.editingId).toBeNull() + expect(edit).toHaveBeenNthCalledWith(2, "selected", {text: "new"}) + expect(server.submit).not.toHaveBeenCalled() + expect(server.remove).not.toHaveBeenCalled() + expect(sendQueued).not.toHaveBeenCalled() + }) + + it("does not submit a new message if the durable row leaves the queue during editing", async () => { + const server: ServerQueueAdapter = { + capabilities: {queue: true, steer: true}, + busy: true, + queued: [{id: "selected", text: "old", source: "server"}], + submit: vi.fn(), + remove: vi.fn(), + edit: vi.fn().mockRejectedValue(new Error("already promoted")), + } + const {result, rerender, sendQueued} = setup({...settledEmpty, server}) + act(() => result.current.beginEdit("selected", "draft")) + rerender({...settledEmpty, server: {...server, queued: []}}) + await act(async () => { + await expect(result.current.commitEdit({text: "new"})).rejects.toThrow( + "already promoted", + ) + }) + expect(result.current.editingId).toBe("selected") + expect(server.submit).not.toHaveBeenCalled() + expect(sendQueued).not.toHaveBeenCalled() + let restored = "" + act(() => { + restored = result.current.cancelEdit() + }) + expect(restored).toBe("draft") + }) +}) + +it.each([false, true])( + "does not overwrite a newer edit when an older save settles (failure=%s)", + async (failure) => { + let resolve!: () => void + let reject!: (error: Error) => void + const edit = vi.fn( + () => + new Promise((yes, no) => { + resolve = yes + reject = no + }), + ) + const server: ServerQueueAdapter = { + capabilities: {queue: true, steer: true}, + busy: true, + queued: [ + {id: "first", text: "old", source: "server"}, + {id: "second", text: "other", source: "server"}, + ], + submit: vi.fn(), + remove: vi.fn(), + edit, + } + const {result} = setup({...settledEmpty, server}) + act(() => result.current.beginEdit("first", "original draft")) + let saving!: string | Promise + act(() => { + saving = result.current.commitEdit({text: "changed"}) + }) + act(() => result.current.beginEdit("second", "new draft")) + await act(async () => { + if (failure) reject(new Error("old failure")) + else resolve() + expect(await saving).toBe("") + }) + expect(result.current.editingId).toBe("second") + let restored = "" + act(() => { + restored = result.current.cancelEdit() + }) + expect(restored).toBe("new draft") + }, +) + describe("cold session capability admission", () => { it.each([true, false])( "waits for queue=%s before choosing the first send owner", @@ -965,3 +1073,174 @@ it("refuses cold Steer if the session settles while capabilities resolve", async expect(server.submit).not.toHaveBeenCalled() expect(view.sendQueued).not.toHaveBeenCalled() }) + +it("holds local-to-server migration while its row is being edited", async () => { + const props = {...settledEmpty, status: "streaming"} + const {result, rerender} = setup(props) + act(() => result.current.submit({text: "old"})) + const id = result.current.queued[0].id + act(() => result.current.beginEdit(id, "draft")) + const server: ServerQueueAdapter = { + capabilities: {queue: true, steer: true}, + busy: true, + queued: [], + submit: vi.fn().mockResolvedValue(undefined), + remove: vi.fn(), + edit: vi.fn(), + } + rerender({...props, server, continuationExecutionId: "continuation"}) + expect(server.submit).not.toHaveBeenCalled() + await act(async () => { + await result.current.commitEdit({text: "edited before migration"}) + }) + expect(server.submit).toHaveBeenCalledOnce() + expect(server.submit).toHaveBeenCalledWith( + expect.objectContaining({id, text: "edited before migration"}), + "queue", + ) +}) + +it.each(["pending", "accepted", "promoted"] as const)( + "keeps same-row durable editing when migration is %s and snapshot lags", + async (state) => { + const props = {...settledEmpty, status: "streaming"} + const {result, rerender, sendQueued} = setup(props) + act(() => result.current.submit({text: "old"})) + const id = result.current.queued[0].id + let accept!: () => void + const server: ServerQueueAdapter = { + capabilities: {queue: true, steer: true}, + busy: true, + queued: [], + submit: vi.fn( + () => + new Promise((resolve) => { + accept = resolve + }), + ), + remove: vi.fn(), + edit: + state === "promoted" + ? vi.fn().mockRejectedValue(new Error("already promoted")) + : vi.fn().mockResolvedValue(undefined), + } + rerender({...props, server, continuationExecutionId: "continuation"}) + expect(server.submit).toHaveBeenCalledOnce() + act(() => result.current.beginEdit(id, "original draft")) + if (state !== "pending") + await act(async () => { + accept() + await Promise.resolve() + }) + let saving!: string | Promise + act(() => { + saving = result.current.commitEdit({text: "corrected"}) + }) + if (state === "pending") { + expect(server.edit).not.toHaveBeenCalled() + await act(async () => { + accept() + expect(await saving).toBe("original draft") + }) + } else if (state === "promoted") { + await act(async () => { + await expect(saving).rejects.toThrow("already promoted") + }) + expect(result.current.editingId).toBe(id) + } else { + await act(async () => { + expect(await saving).toBe("original draft") + }) + } + expect(server.edit).toHaveBeenCalledWith(id, {text: "corrected"}) + expect(server.submit).toHaveBeenCalledOnce() + expect(sendQueued).not.toHaveBeenCalled() + }, +) + +it("retains observed server ownership after a failed edit and a later missing snapshot row", async () => { + const props = {...settledEmpty, status: "streaming"} + const {result, rerender} = setup(props) + act(() => result.current.submit({text: "old"})) + const id = result.current.queued[0].id + act(() => result.current.beginEdit(id, "draft")) + const server: ServerQueueAdapter = { + capabilities: {queue: true, steer: true}, + busy: true, + queued: [{id, text: "old", source: "server"}], + submit: vi.fn(), + remove: vi.fn(), + edit: vi + .fn() + .mockRejectedValueOnce(new Error("retry")) + .mockRejectedValueOnce(new Error("promoted")), + } + rerender({...props, server}) + await act(async () => { + await expect(result.current.commitEdit({text: "corrected"})).rejects.toThrow("retry") + }) + rerender({...props, server: {...server, queued: []}}) + await act(async () => { + await expect(result.current.commitEdit({text: "corrected"})).rejects.toThrow("promoted") + }) + expect(server.edit).toHaveBeenCalledTimes(2) + expect(server.submit).not.toHaveBeenCalled() + expect(result.current.queued).toEqual([]) + expect(result.current.editingId).toBe(id) +}) + +it.each([false, true])( + "recovers an ambiguous migration before editing (server observed=%s)", + async (observed) => { + const props = {...settledEmpty, status: "streaming"} + const {result, rerender, unmount} = setup(props) + act(() => result.current.submit({text: "original admission"})) + const original = result.current.queued[0] + let reject!: (error: Error) => void + const server: ServerQueueAdapter = { + capabilities: {queue: true, steer: true}, + busy: true, + queued: [], + submit: vi + .fn() + .mockImplementationOnce( + () => + new Promise((_yes, no) => { + reject = no + }), + ) + .mockResolvedValueOnce(undefined), + remove: vi.fn(), + edit: vi.fn().mockResolvedValue(undefined), + } + rerender({...props, server, continuationExecutionId: "continuation"}) + act(() => result.current.beginEdit(original.id, "draft")) + let saving!: string | Promise + act(() => { + saving = result.current.commitEdit({text: "corrected"}) + }) + await act(async () => { + reject(new Error("response lost")) + await expect(saving).rejects.toThrow("response lost") + }) + expect(result.current.editingId).toBe(original.id) + expect(server.edit).not.toHaveBeenCalled() + if (observed) { + rerender({ + ...props, + server: {...server, queued: [{...original, source: "server"}]}, + continuationExecutionId: "continuation", + }) + } + await act(async () => { + expect(await result.current.commitEdit({text: "corrected"})).toBe("draft") + }) + expect(server.submit).toHaveBeenNthCalledWith(1, original, "queue") + if (observed) expect(server.submit).toHaveBeenCalledOnce() + else expect(server.submit).toHaveBeenNthCalledWith(2, original, "queue") + expect(server.edit).toHaveBeenCalledOnce() + expect(server.edit).toHaveBeenCalledWith(original.id, {text: "corrected"}) + expect(result.current.queued).toEqual(observed ? [{...original, source: "server"}] : []) + unmount() + }, +) diff --git a/web/packages/agenta-chat/tests/unit/hooks/useServerSessionInputs.test.ts b/web/packages/agenta-chat/tests/unit/hooks/useServerSessionInputs.test.ts index 337fdaf896..59975ad130 100644 --- a/web/packages/agenta-chat/tests/unit/hooks/useServerSessionInputs.test.ts +++ b/web/packages/agenta-chat/tests/unit/hooks/useServerSessionInputs.test.ts @@ -16,15 +16,21 @@ import {useAgentChatQueue} from "../../../src/hooks/useAgentChatQueue" import type {useComposerAttachments} from "../../../src/hooks/useComposerAttachments" import {useServerSessionInputs} from "../../../src/hooks/useServerSessionInputs" -const {buildAgentRequest, fetchCapabilities, fetchSnapshot, removeInput, sendInputNow} = vi.hoisted( - () => ({ - buildAgentRequest: vi.fn(), - fetchCapabilities: vi.fn(), - fetchSnapshot: vi.fn(), - removeInput: vi.fn(), - sendInputNow: vi.fn(), - }), -) +const { + buildAgentRequest, + fetchCapabilities, + fetchSnapshot, + removeInput, + sendInputNow, + updateInput, +} = vi.hoisted(() => ({ + buildAgentRequest: vi.fn(), + fetchCapabilities: vi.fn(), + fetchSnapshot: vi.fn(), + removeInput: vi.fn(), + sendInputNow: vi.fn(), + updateInput: vi.fn(), +})) vi.mock("@agenta/entities/session", async () => { const {atom} = await import("jotai") @@ -35,6 +41,7 @@ vi.mock("@agenta/entities/session", async () => { fetchSessionSnapshotAtom: atom(null, (_get, _set, sessionId: string) => fetchSnapshot(sessionId), ), + updatePendingSessionInputAtom: atom(null, (_get, _set, params) => updateInput(params)), sendPendingSessionInputNowAtom: atom( null, (_get, _set, params: {sessionId: string; inputId: string}) => sendInputNow(params), @@ -83,6 +90,7 @@ beforeEach(() => { fetchSnapshot.mockReset() removeInput.mockReset() sendInputNow.mockReset() + updateInput.mockReset() fetchMock.mockReset() }) @@ -663,3 +671,50 @@ describe("selected queued input Send Now", () => { expect(sendInputNow).not.toHaveBeenCalled() }) }) + +describe("durable queued input editing", () => { + it("patches only the chosen row text and new attachments, then reloads the shared snapshot", async () => { + fetchSnapshot.mockResolvedValue(runningSnapshot()) + updateInput.mockResolvedValue(true) + const {result} = renderHook(() => + useServerSessionInputs({ + entityId: "revision-1", + sessionId: "session-1", + messages: [], + locallyBusy: true, + }), + ) + await waitFor(() => expect(result.current.capabilities.queue).toBe(true)) + await act(() => + result.current.edit("row-2", { + text: "corrected", + fileParts: [ + { + type: "file", + url: "https://files.test/new.pdf", + mediaType: "application/pdf", + filename: "new.pdf", + providerMetadata: {agenta: {attachmentId: "attachment-1"}}, + }, + ], + }), + ) + expect(updateInput).toHaveBeenCalledWith({ + sessionId: "session-1", + inputId: "row-2", + text: "corrected", + attachments: [ + { + uri: "https://files.test/new.pdf", + mime_type: "application/pdf", + filename: "new.pdf", + attachment_id: "attachment-1", + }, + ], + }) + expect(fetchSnapshot.mock.calls.length).toBeGreaterThan(1) + expect(removeInput).not.toHaveBeenCalled() + expect(buildAgentRequest).not.toHaveBeenCalled() + expect(fetchMock).not.toHaveBeenCalled() + }) +}) diff --git a/web/packages/agenta-entities/src/session/api/api.ts b/web/packages/agenta-entities/src/session/api/api.ts index e93a4fb0b3..6d8ebfdd10 100644 --- a/web/packages/agenta-entities/src/session/api/api.ts +++ b/web/packages/agenta-entities/src/session/api/api.ts @@ -14,6 +14,7 @@ import {safeParseWithLogging} from "../../shared/utils/zodSchema" import { mountFileContentResponseSchema, pendingInputAdmissionResponseSchema, + pendingInputResponseSchema, mountFileListResponseSchema, sessionInteractionResponseSchema, sessionInteractionsResponseSchema, @@ -196,6 +197,37 @@ export async function removePendingSessionInput({ return !!data } +export async function updatePendingSessionInput({ + sessionId, + projectId, + appId, + abortSignal, + inputId, + text, + attachments, +}: SessionScopedParams & { + inputId: string + text: string + attachments?: { + uri: string + mime_type: string + filename?: string + attachment_id?: string + }[] +}): Promise { + if (!projectId || !sessionId || !inputId) return false + const data = await callFern("[updatePendingSessionInput]", () => + getSessionsClient().updatePendingSessionInput( + {session_id: sessionId, input_id: inputId, text, attachments}, + projectScopedRequest(projectId, appId, abortSignal), + ), + ) + return ( + safeParseWithLogging(pendingInputResponseSchema, data, "[updatePendingSessionInput]") !== + null + ) +} + export async function sendPendingSessionInputNow({ sessionId, projectId, diff --git a/web/packages/agenta-entities/src/session/core/schema.ts b/web/packages/agenta-entities/src/session/core/schema.ts index fe0be143a2..7cc6a2d8b6 100644 --- a/web/packages/agenta-entities/src/session/core/schema.ts +++ b/web/packages/agenta-entities/src/session/core/schema.ts @@ -265,6 +265,10 @@ export const pendingSessionInputSchema = z.object({ promoted_execution_id: z.string().nullish(), }) +export const pendingInputResponseSchema = z.object({ + input: pendingSessionInputSchema, +}) + export const pendingInputAdmissionResponseSchema = z.object({ action: z.enum(["execute", "pending"]), input: pendingSessionInputSchema.nullish(), diff --git a/web/packages/agenta-entities/src/session/index.ts b/web/packages/agenta-entities/src/session/index.ts index b04a91cfde..5b1189e978 100644 --- a/web/packages/agenta-entities/src/session/index.ts +++ b/web/packages/agenta-entities/src/session/index.ts @@ -23,6 +23,7 @@ export { fetchSessionDurableApprovalsCapability, removePendingSessionInput, sendPendingSessionInputNow, + updatePendingSessionInput, invalidateSessionDurableApprovalsCapability, commandSessionStream, cancelSessionExecution, @@ -110,6 +111,7 @@ export { fetchSessionSnapshotAtom, removePendingSessionInputAtom, sendPendingSessionInputNowAtom, + updatePendingSessionInputAtom, } from "./state/pendingInputs" export { deriveStreamNest, diff --git a/web/packages/agenta-entities/src/session/state/pendingInputs.ts b/web/packages/agenta-entities/src/session/state/pendingInputs.ts index d9ce53c727..a37aef415b 100644 --- a/web/packages/agenta-entities/src/session/state/pendingInputs.ts +++ b/web/packages/agenta-entities/src/session/state/pendingInputs.ts @@ -6,6 +6,7 @@ import { fetchSessionSnapshot, removePendingSessionInput, sendPendingSessionInputNow, + updatePendingSessionInput, } from "../api/api" export const fetchSessionCapabilitiesAtom = atom(null, async (get, _set, sessionId: string) => { @@ -33,3 +34,25 @@ export const sendPendingSessionInputNowAtom = atom( return sendPendingSessionInputNow({projectId, ...params}) }, ) + +export const updatePendingSessionInputAtom = atom( + null, + async ( + get, + _set, + params: { + sessionId: string + inputId: string + text: string + attachments?: { + uri: string + mime_type: string + filename?: string + attachment_id?: string + }[] + }, + ) => { + const projectId = get(projectIdAtom) ?? "" + return updatePendingSessionInput({projectId, ...params}) + }, +) diff --git a/web/packages/agenta-entities/tests/unit/session-continuation-resume-api.test.ts b/web/packages/agenta-entities/tests/unit/session-continuation-resume-api.test.ts index 0cb08d4ca3..92e946afe9 100644 --- a/web/packages/agenta-entities/tests/unit/session-continuation-resume-api.test.ts +++ b/web/packages/agenta-entities/tests/unit/session-continuation-resume-api.test.ts @@ -1,16 +1,18 @@ import type {SessionCapabilities, SessionStreamResponse} from "@agentaai/api-client" import {beforeEach, describe, expect, expectTypeOf, it, vi} from "vitest" -const {resume, fetchStream, sendNow} = vi.hoisted(() => ({ +const {resume, fetchStream, sendNow, updateInput} = vi.hoisted(() => ({ resume: vi.fn(), fetchStream: vi.fn(), sendNow: vi.fn(), + updateInput: vi.fn(), })) vi.mock("@agenta/sdk/resources", () => ({ getSessionsClient: () => ({ resumeSessionContinuation: resume, sendPendingSessionInputNow: sendNow, + updatePendingSessionInput: updateInput, fetchSessionStream: fetchStream, }), getLowPrioritySessionsClient: vi.fn(), @@ -20,6 +22,7 @@ vi.mock("@agenta/sdk/resources", () => ({ import { fetchSessionCapabilities, + updatePendingSessionInput, sendPendingSessionInputNow, fetchSessionDurableApprovalsCapability, invalidateSessionDurableApprovalsCapability, @@ -201,3 +204,34 @@ it.each([ sendPendingSessionInputNow({projectId: "project", sessionId: "session", inputId: "input"}), ).resolves.toBe(accepted) }) + +it.each([ + [ + { + input: { + id: "input", + session_id: "session", + content: {data: {inputs: {messages: []}}}, + position: 1, + state: "pending", + policy: "queue", + }, + }, + true, + ], + [{}, false], + [{input: null}, false], + [{input: {id: "input"}}, false], + [{action: "pending"}, false], + [null, false], +])("validates a queued edit receipt %j", async (response, accepted) => { + updateInput.mockResolvedValue(response) + await expect( + updatePendingSessionInput({ + projectId: "project", + sessionId: "session", + inputId: "input", + text: "edited", + }), + ).resolves.toBe(accepted) +}) diff --git a/web/storybook/stories/presentational/QueuedMessagesDock.stories.tsx b/web/storybook/stories/presentational/QueuedMessagesDock.stories.tsx index cde60e8d31..bcbbe302ca 100644 --- a/web/storybook/stories/presentational/QueuedMessagesDock.stories.tsx +++ b/web/storybook/stories/presentational/QueuedMessagesDock.stories.tsx @@ -84,11 +84,11 @@ export const ServerBacked: Story = { touch editable initial={[ - {...THREE[0], source: "server", editable: false, policy: "steer"}, + {...THREE[0], source: "server", editable: true, policy: "steer"}, { ...THREE[1], source: "server", - editable: false, + editable: true, policy: "queue", attachmentCount: 1, }, @@ -168,3 +168,17 @@ export const SendNowFailure: Story = { /> ), } + +export const EditedRowNoLongerQueued: Story = { + render: () => ( + + + + ), +}