diff --git a/docs/convergence/protocol-registry.md b/docs/convergence/protocol-registry.md index f851289a..44ad1804 100644 --- a/docs/convergence/protocol-registry.md +++ b/docs/convergence/protocol-registry.md @@ -7,7 +7,7 @@ This document is generated from [`protocols/registry.json`](../../protocols/regi | Family | Class | Owner | Active IDs | Persisted/exported | | --- | --- | --- | ---: | --- | -| [activity-event-projection](#activity-event-projection) | `projection` | Run activity and HTTP maintainers | 5 | true | +| [activity-event-projection](#activity-event-projection) | `projection` | Run activity and HTTP maintainers | 6 | true | | [agent-browser-supervisor-ledger](#agent-browser-supervisor-ledger) | `internal-durable` | Agent browser, Supervisor tool, and persistence maintainers | 11 | true | | [agent-scheduling-delivery-ledgers](#agent-scheduling-delivery-ledgers) | `internal-durable` | Agent graph, scheduler, and batch-delivery maintainers | 92 | true | | [analyzer-interchange](#analyzer-interchange) | `external-durable` | Analyzer contract maintainers | 40 | true | @@ -66,13 +66,14 @@ This document is generated from [`protocols/registry.json`](../../protocols/regi - Readers: - `activity-event-projection-reader` (`v1`, active) at `internal/runactivity` -
5 active identifiers +
6 active identifiers - `operator_action_center.v1` - `operator_action_item.v1` - `run-event-poll.v1` - `run-events.v1` - `run_activity.v1` +- `thread_reply_updates.v1`
diff --git a/docs/evidence/chat-unread/README.md b/docs/evidence/chat-unread/README.md new file mode 100644 index 00000000..09b022ac --- /dev/null +++ b/docs/evidence/chat-unread/README.md @@ -0,0 +1,45 @@ +# Conversation unread markers + +Captured on Windows in Chrome with the production `cmd/workbenchjourney` API, +SQLite store, offline model provider, and a built `web/dist`. Replies came from +real workspace-read executions. No browser API mocks or DOM edits were used. + +- Before: commit `52c7da77fd83251b61d1696735a3d98f521ce37f`, UI digest + `78d6b88f6c424bb8718cf4779e7a4fc6f3cc24cd7c861580b0471563a14ddea6`. +- After: this change, UI digest + `ee1de139b321c223b4c6d8e17ecb6d5cb80a11b39b3bf516516b76d57f9ec07b`. +- Desktop viewport: 1440 × 1000. Narrow viewport: 390 × 844. + +| Image | Observed behavior | +| --- | --- | +| [Before](before-unread.png) | A background conversation has a saved reply but no unread marker. | +| [Background reply](after-background-reply.png) | A new saved reply in the other conversation adds a dot before its title. The existing execution-state dot stays at the right. | +| [Saved-message history](after-history-retains-unread.png) | Opening an earlier saved message retains the unread marker. Returning to latest then clears it. | +| [Manual unread](after-manual-unread.png) | Marking the current conversation unread keeps the dot and offers “Mark as read” in its menu. | +| [Narrow sidebar](after-narrow-sidebar.png) | A new reply in the current conversation stays unread while the sidebar covers its content. Opening that conversation clears it after the latest view is available. | + +The two background replies had identical text but distinct saved message IDs: +`evt-20261021191411-57d288cf6c25` and +`evt-20261021191519-4bf5a1b6b80c`. The second reply still produced an unread marker. +Manual unread also survived page reload and reconnect; clicking the current +sidebar row again cleared it. Explicit “Mark as read” cleared the marker. +Both production pages reported zero browser console errors and warnings. + +Focused React checks passed for the unread store, reading confirmation, +transcript refresh, sidebar, conversation menu, app integration, and conversation +behavior (173 tests, followed by all 7 unread-store tests after adding the +malformed-record case). `npm run typecheck`, `npm run build`, `npm run check:api`, +`git diff --check`, and the protocol registry check against the backend base +passed. The build retained the existing large-chunk warning. + +Tests cover failed fresh-head reads, replies arriving during a read, old pending +identities outside the latest window, empty feed continuation pages, persistence +failure, and data-store changes. Focus/visibility cancellation is covered by +these tests. Headless Chrome kept both tabs focused and visible, so tab switching +was not used as evidence for actual window-focus changes. + +Final integration also includes the sidebar-width and notification changes from +its PR base. The focused combined suite passed 93 tests across seven files and +the production build passed. A navigation-acceptance regression ensures an +unaccepted file-editor leave request does not clear the target chat's manual +unread state; explicitly reopening the current route still acknowledges it. diff --git a/docs/evidence/chat-unread/after-background-reply.png b/docs/evidence/chat-unread/after-background-reply.png new file mode 100644 index 00000000..fe661c50 Binary files /dev/null and b/docs/evidence/chat-unread/after-background-reply.png differ diff --git a/docs/evidence/chat-unread/after-history-retains-unread.png b/docs/evidence/chat-unread/after-history-retains-unread.png new file mode 100644 index 00000000..6b3bbcc1 Binary files /dev/null and b/docs/evidence/chat-unread/after-history-retains-unread.png differ diff --git a/docs/evidence/chat-unread/after-manual-unread.png b/docs/evidence/chat-unread/after-manual-unread.png new file mode 100644 index 00000000..0b21ce5f Binary files /dev/null and b/docs/evidence/chat-unread/after-manual-unread.png differ diff --git a/docs/evidence/chat-unread/after-narrow-sidebar.png b/docs/evidence/chat-unread/after-narrow-sidebar.png new file mode 100644 index 00000000..14764db2 Binary files /dev/null and b/docs/evidence/chat-unread/after-narrow-sidebar.png differ diff --git a/docs/evidence/chat-unread/before-unread.png b/docs/evidence/chat-unread/before-unread.png new file mode 100644 index 00000000..eb8e9627 Binary files /dev/null and b/docs/evidence/chat-unread/before-unread.png differ diff --git a/docs/openapi.json b/docs/openapi.json index 55c7d236..bc9d0939 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -27635,6 +27635,96 @@ "x-cyberagent-read-only": true } }, + "/api/v1/threads/reply-updates": { + "get": { + "operationId": "listThreadReplyUpdates", + "summary": "Observe new saved public replies across conversations", + "description": "Read-only incremental observation of saved public assistant message identities in active and archived conversations. Omitting cursor establishes the current baseline with no historical items. Subsequent requests scan at most limit source events in global save order; an empty items array can still have has_more=true. Continue using data.cursor until has_more=false, then retain that cursor for the next poll. Cursor scope includes the exact data_store_id. Persist returned message identities before the cursor; an error returns no partial cursor. Replaying a cursor preserves message identities. Separate saved messages with equal text remain separate updates. Excludes deleted conversations, copied branch history, users, tools, hidden/internal events and provisional streaming bytes. No message body or private event payload is returned.", + "tags": [ + "Threads" + ], + "parameters": [ + { + "name": "limit", + "in": "query", + "description": "Maximum source events scanned, including filtered events; not a requested number of visible replies", + "schema": { + "default": 50, + "maximum": 100, + "minimum": 1, + "type": "integer" + } + }, + { + "name": "cursor", + "in": "query", + "description": "Opaque cursor from data.cursor, bound to the exact data store; omit only to establish a new baseline", + "schema": { + "maxLength": 512, + "minLength": 1, + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Successful read", + "content": { + "application/json": { + "schema": { + "additionalProperties": false, + "properties": { + "data": { + "$ref": "#/components/schemas/ThreadReplyUpdatesView" + }, + "request_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "version": { + "const": "api.v1", + "type": "string" + } + }, + "required": [ + "version", + "request_id", + "data" + ], + "type": "object" + } + } + } + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "403": { + "$ref": "#/components/responses/Forbidden" + }, + "414": { + "$ref": "#/components/responses/RequestTooLarge" + }, + "429": { + "$ref": "#/components/responses/ResourceExhausted" + }, + "500": { + "$ref": "#/components/responses/InternalError" + }, + "503": { + "$ref": "#/components/responses/Unavailable" + }, + "504": { + "$ref": "#/components/responses/GatewayTimeout" + } + }, + "x-cyberagent-read-only": true + } + }, "/api/v1/threads/search": { "get": { "operationId": "searchSavedThreadMessages", @@ -72455,6 +72545,91 @@ ], "type": "object" }, + "ThreadReplyUpdateView": { + "additionalProperties": false, + "properties": { + "created_at": { + "format": "date-time", + "type": "string" + }, + "message_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "run_ordinal": { + "format": "int64", + "minimum": 1, + "type": "integer" + }, + "sequence": { + "format": "int64", + "minimum": 1, + "type": "integer" + }, + "thread_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "thread_status": { + "enum": [ + "active", + "archived" + ], + "type": "string" + } + }, + "required": [ + "thread_id", + "message_id", + "thread_status", + "run_ordinal", + "sequence", + "created_at" + ], + "type": "object" + }, + "ThreadReplyUpdatesView": { + "additionalProperties": false, + "properties": { + "cursor": { + "maxLength": 512, + "minLength": 1, + "type": "string" + }, + "data_store_id": { + "maxLength": 256, + "minLength": 1, + "pattern": "^ds1_[0-9a-f]{64}$", + "type": "string" + }, + "has_more": { + "type": "boolean" + }, + "items": { + "items": { + "$ref": "#/components/schemas/ThreadReplyUpdateView" + }, + "maxItems": 100, + "type": "array" + }, + "version": { + "enum": [ + "thread_reply_updates.v1" + ], + "type": "string" + } + }, + "required": [ + "version", + "data_store_id", + "cursor", + "has_more", + "items" + ], + "type": "object" + }, "ThreadRequestObservationView": { "additionalProperties": false, "properties": { diff --git a/docs/thread-reply-updates.md b/docs/thread-reply-updates.md new file mode 100644 index 00000000..bb64f7ad --- /dev/null +++ b/docs/thread-reply-updates.md @@ -0,0 +1,86 @@ +# Saved reply updates + +`GET /api/v1/threads/reply-updates` uses the existing read bearer and GET-only +HTTP boundary. It observes active and archived conversations without changing +their lifecycle. Deleted conversations and copied branch history are excluded. + +The response uses the normal `api.v1` envelope. Its `data` has this shape: + +```json +{ + "version": "thread_reply_updates.v1", + "data_store_id": "ds1_…", + "cursor": "opaque-cursor", + "has_more": false, + "items": [ + { + "thread_id": "thread-…", + "message_id": "event-…", + "thread_status": "active", + "run_ordinal": 2, + "sequence": 18, + "created_at": "2026-10-22T10:00:00Z" + } + ] +} +``` + +- Omit `cursor` on first use. The server returns the current position and an + empty `items` array, so enabling observation does not mark all history unread. + An explicitly empty cursor is invalid. +- Supply the returned `data.cursor` on every later poll. This is not a + `page.next_cursor`. An idle poll can return the same cursor. +- `limit` defaults to 50 and accepts 1–100. It bounds **source events scanned**, + including filtered events. The store reads one extra event to detect more work. + `items: []` with `has_more: true` is a valid page: continue from its cursor. +- Each request captures the current highest saved event position before scanning. + `has_more: false` means it has covered all source events through that position. + Events appended after that capture remain available to the next poll. +- Items are ordered by global event save position. Appending events or creating + a successor Run does not shift a continuation. `(run_ordinal, sequence)` gives + the item's position in its Thread transcript; `created_at` is not a cursor. +- Each item identifies a new saved public assistant message or public progress + update accepted by the existing Go transcript projection. User input, queued + or cancelled input, tool output, internal events and provisional streaming + bytes cannot produce items. Neither message bodies nor raw event data appear + in the response. +- Equal text saved as separate messages retains separate identities. This feed + does not reproduce the conversation renderer's text merging. Replaying a + cursor preserves message IDs, so consumers can apply results idempotently. +- Persist the returned identities/read-state changes before persisting their + cursor. A failed request returns no partial items or advanced cursor; retry the + previous cursor. Do not discard an unread marker merely because a poll fails. +- Cursors belong to one `data_store_id`. Another database, a replacement, or a + copied database at a different path rejects the old cursor. Keep local state + under the existing connection/data-store namespace; do not apply an in-flight + response after switching that namespace. A new namespace starts with its own + baseline. Switching back can resume the original namespace's saved cursor. + +The OpenAPI document and generated TypeScript schema define the exact field +bounds. This endpoint supplies observations; the frontend owns the local read +position and explicit manual-unread preference. + +## Local unread markers + +The sidebar marks conversations with new saved replies and offers **Mark as +read** / **Mark as unread** in each conversation's menu. The cursor, pending +message identities, and manual preference are saved together in the existing +recovery namespace for the browser origin, API URL, and data-store identity. +Polling follows empty continuation pages and retries the unchanged checkpoint +after a request failure. A failed local write leaves the previous complete +checkpoint available after reload. + +Automatic reading requires the conversation's latest view, a visible and focused +window, and a scroll position at the bottom. Saved-message history, inspectors, +an open narrow-screen sidebar, and covered conversation content do not acknowledge +replies. Visible saved assistant identities are acknowledged directly. On entering +the latest view, the client first catches up the reply feed, freezes the pending +identities, and requests a fresh transcript head. Only a successful read rendered +in the same reading context acknowledges that frozen range, including older +pending messages outside the latest window. Replies arriving during the request +remain pending until observed or included in a later confirmed range. + +A manual unread marker survives polling, automatic reading, and reload. Explicitly +opening the conversation from the sidebar, including clicking the current row +again, clears that manual preference. **Mark as read** clears both the manual +marker and currently known pending replies. diff --git a/internal/httpapi/openapi.go b/internal/httpapi/openapi.go index 4f518fd0..406fca61 100644 --- a/internal/httpapi/openapi.go +++ b/internal/httpapi/openapi.go @@ -1072,6 +1072,16 @@ func openAPIOperationSpecs() []openAPIOperationSpec { "minLength": domain.MinAgentOperationKeyBytes, "maxLength": domain.MaxAgentOperationKeyBytes, "pattern": `^\S+$`}}, }}, + {Path: ThreadReplyUpdatesPath, OperationID: "listThreadReplyUpdates", + Summary: "Observe new saved public replies across conversations", Tag: "Threads", + Description: "Read-only incremental observation of saved public assistant message identities in active and archived conversations. Omitting cursor establishes the current baseline with no historical items. Subsequent requests scan at most limit source events in global save order; an empty items array can still have has_more=true. Continue using data.cursor until has_more=false, then retain that cursor for the next poll. Cursor scope includes the exact data_store_id. Persist returned message identities before the cursor; an error returns no partial cursor. Replaying a cursor preserves message identities. Separate saved messages with equal text remain separate updates. Excludes deleted conversations, copied branch history, users, tools, hidden/internal events and provisional streaming bytes. No message body or private event payload is returned.", + DataType: reflect.TypeOf(ThreadReplyUpdatesView{}), + Parameters: []openAPIParameter{ + {Name: "limit", In: "query", Description: "Maximum source events scanned, including filtered events; not a requested number of visible replies", + Schema: map[string]any{"type": "integer", "minimum": 1, "maximum": MaxPageLimit, "default": DefaultPageLimit}}, + {Name: "cursor", In: "query", Description: "Opaque cursor from data.cursor, bound to the exact data store; omit only to establish a new baseline", + Schema: map[string]any{"type": "string", "minLength": 1, "maxLength": MaxCursorBytes}}, + }}, {Path: "/api/v1/threads/search", OperationID: "searchSavedThreadMessages", Summary: "Search saved conversation titles and public message bodies", Tag: "Threads", Description: "Searches complete saved user and assistant text, including compacted history, public model progress and inherited branch replies, before pagination. With thread_id, searches only message bodies in that exact conversation; otherwise also searches conversation titles. Uses literal ASCII case-insensitive substrings, with exact matching for other Unicode. Redacts and checks the existing public transcript projection before returning a match. Excludes deleted conversations, tools, peer messages and internal events. Defaults to active conversations; include_archived also includes read-only archived history. Results use creation-time and identity cursors scoped to the normalized query, Thread and archive filter.", @@ -2871,6 +2881,19 @@ func jsonField(field reflect.StructField) (string, bool, bool) { } func applyOpenAPIFieldMetadata(typeName string, fieldName string, schema map[string]any) { + if typeName == "ThreadReplyUpdatesView" { + switch fieldName { + case "data_store_id": + schema["pattern"], schema["minLength"], schema["maxLength"] = "^ds1_[0-9a-f]{64}$", 68, 68 + case "cursor": + schema["minLength"], schema["maxLength"] = 1, MaxCursorBytes + case "items": + schema["maxItems"] = MaxPageLimit + } + } + if typeName == "ThreadReplyUpdateView" && (fieldName == "run_ordinal" || fieldName == "sequence") { + schema["minimum"] = 1 + } if typeName == "MemoryCandidate" && fieldName == "fingerprint" || typeName == "MemoryCandidateSource" && fieldName == "completion_fingerprint" || (typeName == "memoryCandidateConfirmRequestView" || typeName == "memoryCandidateRejectRequestView") && fieldName == "expected_fingerprint" { @@ -3489,6 +3512,8 @@ var openAPIFieldEnums = map[string][]string{ "RunActivityView.version": {runactivity.ProtocolVersion}, "RunActivityItemView.kind": {string(runactivity.KindHarnessStatus), string(runactivity.KindModelUpdate), string(runactivity.KindOperatorInput), string(runactivity.KindModelCall), string(runactivity.KindToolCall), string(runactivity.KindApproval), string(runactivity.KindFileChange), string(runactivity.KindPlan), string(runactivity.KindDependency), string(runactivity.KindBrowser)}, "RunActivityItemView.source": {string(runactivity.SourceHarness), string(runactivity.SourceModel), string(runactivity.SourceOperator)}, + "ThreadReplyUpdatesView.version": {ThreadReplyUpdatesVersion}, + "ThreadReplyUpdateView.thread_status": {"active", "archived"}, "ThreadTranscriptItemView.version": {threadtranscript.ProtocolVersion}, "ThreadTranscriptItemView.activity_type": {string(threadtranscript.TypeMessage), string(threadtranscript.TypeSearch), string(threadtranscript.TypeRead), string(threadtranscript.TypeEdit), string(threadtranscript.TypeExecute), string(threadtranscript.TypeVerify), string(threadtranscript.TypeApproval), string(threadtranscript.TypeCheckpoint), string(threadtranscript.TypeDelivery)}, "ThreadTranscriptItemView.stage": {string(threadtranscript.StageStarted), string(threadtranscript.StageArgumentsReady), string(threadtranscript.StageRunning), string(threadtranscript.StageResult), string(threadtranscript.StageBlocked)}, diff --git a/internal/httpapi/thread_control.go b/internal/httpapi/thread_control.go index 3a7a72b1..4c31c7c3 100644 --- a/internal/httpapi/thread_control.go +++ b/internal/httpapi/thread_control.go @@ -118,6 +118,9 @@ func (a *API) routeThreads(request *http.Request, segments []string) (any, *Page case 1: return a.threads(request) case 2: + if segments[1] == "reply-updates" { + return a.threadReplyUpdates(request) + } if segments[1] == "search" { return a.searchThreadMessages(request) } diff --git a/internal/httpapi/thread_reply_updates.go b/internal/httpapi/thread_reply_updates.go new file mode 100644 index 00000000..f6fb4019 --- /dev/null +++ b/internal/httpapi/thread_reply_updates.go @@ -0,0 +1,72 @@ +package httpapi + +import ( + "context" + "net/http" + "strconv" + "time" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/domain" +) + +const ( + ThreadReplyUpdatesPath = "/api/v1/threads/reply-updates" + ThreadReplyUpdatesVersion = "thread_reply_updates.v1" +) + +type threadReplyUpdateStore interface { + DataStoreID() string + ListThreadReplyUpdates(context.Context, string, int) ([]domain.ThreadSearchMatch, string, bool, error) +} + +type ThreadReplyUpdateView struct { + ThreadID string `json:"thread_id"` + MessageID string `json:"message_id"` + ThreadStatus string `json:"thread_status"` + RunOrdinal int64 `json:"run_ordinal"` + Sequence int64 `json:"sequence"` + CreatedAt time.Time `json:"created_at"` +} + +type ThreadReplyUpdatesView struct { + Version string `json:"version"` + DataStoreID string `json:"data_store_id"` + Cursor string `json:"cursor"` + HasMore bool `json:"has_more"` + Items []ThreadReplyUpdateView `json:"items"` +} + +func (a *API) threadReplyUpdates(request *http.Request) (any, *Page, error) { + values := request.URL.Query() + if err := validateSingleQueryValues(values, "cursor", "limit"); err != nil { + return nil, nil, err + } + limit := DefaultPageLimit + if raw, exists := singleQueryValue(values, "limit"); exists { + var err error + limit, err = strconv.Atoi(raw) + if err != nil || limit < 1 || limit > MaxPageLimit { + return nil, nil, apperror.New(apperror.CodeInvalidArgument, "reply update limit must be between 1 and 100") + } + } + cursor, present := singleQueryValue(values, "cursor") + if present && cursor == "" { + return nil, nil, apperror.New(apperror.CodeInvalidArgument, "reply update cursor must not be empty") + } + reader, ok := a.store.(threadReplyUpdateStore) + if !ok || reader.DataStoreID() == "" { + return nil, nil, apperror.New(apperror.CodeUnavailable, "reply updates require a stable data store identity") + } + items, next, more, err := reader.ListThreadReplyUpdates(request.Context(), cursor, limit) + if err != nil { + return nil, nil, err + } + view := ThreadReplyUpdatesView{Version: ThreadReplyUpdatesVersion, DataStoreID: reader.DataStoreID(), + Cursor: next, HasMore: more, Items: make([]ThreadReplyUpdateView, 0, len(items))} + for _, item := range items { + view.Items = append(view.Items, ThreadReplyUpdateView{ThreadID: item.ThreadID, MessageID: item.MessageID, + ThreadStatus: string(item.ThreadStatus), RunOrdinal: item.RunOrdinal, Sequence: item.Sequence, CreatedAt: item.CreatedAt}) + } + return view, nil, nil +} diff --git a/internal/httpapi/thread_reply_updates_test.go b/internal/httpapi/thread_reply_updates_test.go new file mode 100644 index 00000000..b0cdb5ca --- /dev/null +++ b/internal/httpapi/thread_reply_updates_test.go @@ -0,0 +1,93 @@ +package httpapi + +import ( + "context" + "errors" + "net/http" + "net/url" + "reflect" + "strings" + "testing" + + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/session" +) + +func TestThreadReplyUpdatesHTTPBaselinePublicIDsAndReadAuthorization(t *testing.T) { + f := newAPIFixture(t) + if _, err := f.store.SaveSessionMessage(t.Context(), session.NewMessage(f.run.SessionID, "assistant", "old reply")); err != nil { + t.Fatal(err) + } + var baseline ThreadReplyUpdatesView + decodeData(t, f.get(t, ThreadReplyUpdatesPath), &baseline) + if baseline.Version != ThreadReplyUpdatesVersion || baseline.DataStoreID != f.store.DataStoreID() || + baseline.Cursor == "" || baseline.Items == nil || len(baseline.Items) != 0 || baseline.HasMore { + t.Fatalf("baseline contract: %#v", baseline) + } + for _, message := range []session.Message{ + session.NewMessage(f.run.SessionID, "system", "HiddenReplyUpdateCanary"), + session.NewMessage(f.run.SessionID, "assistant", "PUBLIC_BODY_NOT_IN_FEED"), + } { + if _, err := f.store.SaveSessionMessage(t.Context(), message); err != nil { + t.Fatal(err) + } + } + path := ThreadReplyUpdatesPath + "?limit=1&cursor=" + url.QueryEscape(baseline.Cursor) + var first ThreadReplyUpdatesView + decodeData(t, f.get(t, path), &first) + if len(first.Items) != 0 || !first.HasMore || first.Cursor == baseline.Cursor { + t.Fatalf("empty source page: %#v", first) + } + var reply ThreadReplyUpdatesView + response := f.get(t, ThreadReplyUpdatesPath+"?limit=1&cursor="+url.QueryEscape(first.Cursor)) + decodeData(t, response, &reply) + if len(reply.Items) != 1 || reply.Items[0].ThreadID != domain.InitialThreadID(f.run.ID) || + reply.Items[0].MessageID == "" || reply.Items[0].RunOrdinal != 1 || reply.Items[0].Sequence <= 0 || + reply.Items[0].ThreadStatus != "active" || reply.HasMore { + t.Fatalf("reply identity: %#v", reply) + } + for _, forbidden := range []string{"HiddenReplyUpdateCanary", "PUBLIC_BODY_NOT_IN_FEED", "payload_json", "content_sha256", "subject_id"} { + if strings.Contains(response.Body.String(), forbidden) { + t.Fatalf("feed exposed %s", forbidden) + } + } + var replay ThreadReplyUpdatesView + decodeData(t, f.get(t, ThreadReplyUpdatesPath+"?limit=1&cursor="+url.QueryEscape(first.Cursor)), &replay) + if !reflect.DeepEqual(reply, replay) { + t.Fatal("replay changed message identities") + } + for _, token := range []string{"", testControlToken, "wrong-token"} { + assertAPIError(t, f.request(t, http.MethodGet, path, token, "127.0.0.1:8765", "127.0.0.1:45000", nil), http.StatusUnauthorized, "POLICY_DENIED") + } + assertAPIError(t, f.request(t, http.MethodPost, path, testAccessToken, "127.0.0.1:8765", "127.0.0.1:45000", nil), http.StatusMethodNotAllowed, "INVALID_ARGUMENT") + assertAPIError(t, f.request(t, http.MethodGet, path, testAccessToken, "127.0.0.1:8765", "127.0.0.1:45000", strings.NewReader("body")), http.StatusBadRequest, "INVALID_ARGUMENT") +} + +type unavailableReplyUpdateStore struct { + Store + identity string +} + +func (s unavailableReplyUpdateStore) DataStoreID() string { return s.identity } +func (s unavailableReplyUpdateStore) ListThreadReplyUpdates(context.Context, string, int) ([]domain.ThreadSearchMatch, string, bool, error) { + return []domain.ThreadSearchMatch{{MessageID: "partial-reply"}}, "partial-cursor", true, errors.New("reply reader failed") +} + +func TestThreadReplyUpdatesHTTPValidationStoreIsolationAndFailure(t *testing.T) { + f := newAPIFixture(t) + var baseline ThreadReplyUpdatesView + decodeData(t, f.get(t, ThreadReplyUpdatesPath), &baseline) + for _, query := range []string{"?cursor=", "?cursor=invalid", "?cursor=a&cursor=b", "?limit=0", "?limit=101", "?limit=1&limit=2", "?unknown=yes"} { + assertAPIError(t, f.get(t, ThreadReplyUpdatesPath+query), http.StatusBadRequest, "INVALID_ARGUMENT") + } + other := newAPIFixture(t) + assertAPIError(t, other.get(t, ThreadReplyUpdatesPath+"?cursor="+url.QueryEscape(baseline.Cursor)), http.StatusBadRequest, "INVALID_ARGUMENT") + original := f.api.store + f.api.store = legacyHealthStore{Store: original} + assertAPIError(t, f.get(t, ThreadReplyUpdatesPath), http.StatusServiceUnavailable, "UNAVAILABLE") + f.api.store = unavailableReplyUpdateStore{Store: original, identity: f.store.DataStoreID()} + response := f.get(t, ThreadReplyUpdatesPath+"?cursor="+url.QueryEscape(baseline.Cursor)) + if response.Code < 500 || strings.Contains(response.Body.String(), "partial-cursor") || strings.Contains(response.Body.String(), "partial-reply") { + t.Fatalf("failed reader returned partial state: %d %s", response.Code, response.Body.String()) + } +} diff --git a/internal/store/thread_reply_updates.go b/internal/store/thread_reply_updates.go new file mode 100644 index 00000000..2d7b9c0d --- /dev/null +++ b/internal/store/thread_reply_updates.go @@ -0,0 +1,182 @@ +package store + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "errors" + "io" + "strings" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/events" + "cyberagent-workbench/internal/redact" + "cyberagent-workbench/internal/threadtranscript" +) + +type threadReplyCursor struct { + Version int `json:"v"` + Store string `json:"s"` + After int64 `json:"a"` +} + +type threadReplyRecord struct { + position int64 + thread domain.ThreadSearchMatch + source threadtranscript.Source +} + +// ListThreadReplyUpdates scans at most limit source events plus one lookahead. +// An empty cursor establishes the current baseline without replaying history. +// Active and archived Threads are observed; deleted Threads never yield items. +// Empty result pages may have more work. Persist items before their cursor and +// continue while hasMore is true. Replaying a cursor returns the same message IDs. +// Equal text in separate saved messages does not establish replay identity. +func (s *SQLiteStore) ListThreadReplyUpdates(ctx context.Context, cursor string, limit int) ( + items []domain.ThreadSearchMatch, next string, hasMore bool, err error, +) { + if limit < 1 || limit > 100 { + return nil, "", false, apperror.New(apperror.CodeInvalidArgument, "reply update limit must be between 1 and 100") + } + var head int64 + if err := s.db.QueryRowContext(ctx, `SELECT COALESCE(MAX(id),0) FROM run_events`).Scan(&head); err != nil { + return nil, "", false, err + } + state := threadReplyCursor{Version: 1, Store: s.DataStoreID(), After: head} + items = make([]domain.ThreadSearchMatch, 0) + if cursor == "" { + return items, encodeThreadReplyCursor(state), false, nil + } + state, err = decodeThreadReplyCursor(cursor, s.DataStoreID(), head) + if err != nil { + return nil, "", false, err + } + records, err := s.threadReplyRecords(ctx, state.After, head, limit+1) + if err != nil { + return nil, "", false, err + } + hasMore = len(records) > limit + if hasMore { + records = records[:limit] + } + for _, record := range records { + if record.thread.ThreadStatus == domain.ThreadActive || record.thread.ThreadStatus == domain.ThreadArchived { + visible, projectionErr := publicThreadReplySource(record.thread.ThreadID, record.source) + if projectionErr != nil { + return nil, "", false, projectionErr + } + if visible { + items = append(items, record.thread) + } + } + state.After = record.position + } + return items, encodeThreadReplyCursor(state), hasMore, nil +} + +func (s *SQLiteStore) threadReplyRecords(ctx context.Context, after, head int64, limit int) ([]threadReplyRecord, error) { + // Advance over private and unbound events too, without projecting their data. + // The global primary key keeps additions and Run succession from shifting a page. + rows, err := s.db.QueryContext(ctx, `SELECT event.id, COALESCE(thread.id,''), + COALESCE(thread.status,''), COALESCE(binding.ordinal,0), COALESCE(binding.session_id,''), + COALESCE(run.status,''), event.event_id,event.version,event.run_id,event.mission_id, + event.sequence,event.type,event.source,COALESCE(event.subject_id,''), + CASE WHEN thread.id IS NOT NULL AND event.source<>'thread_branch' + AND event.type IN ('session.message_created','model.public_commentary') + THEN event.payload_json ELSE '{}' END,event.created_at + FROM run_events event + LEFT JOIN thread_runs binding ON binding.run_id=event.run_id + LEFT JOIN threads thread ON thread.id=binding.thread_id AND thread.status IN ('active','archived') + LEFT JOIN runs run ON run.id=event.run_id + WHERE event.id>? AND event.id<=? ORDER BY event.id LIMIT ?`, after, head, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var records []threadReplyRecord + for rows.Next() { + var record threadReplyRecord + var event events.Event + var created string + if err := rows.Scan(&record.position, &record.thread.ThreadID, &record.thread.ThreadStatus, + &record.source.Ordinal, &record.source.SessionID, &record.source.RunStatus, + &event.EventID, &event.Version, &event.RunID, &event.MissionID, &event.Sequence, + &event.Type, &event.Source, &event.SubjectID, &event.PayloadJSON, &created); err != nil { + return nil, err + } + event.CreatedAt = parseTS(created) + record.source.RunID, record.source.Sequence = event.RunID, event.Sequence + record.source.CreatedAt, record.source.Event = event.CreatedAt, &event + record.thread.ID, record.thread.MessageID = event.EventID, event.EventID + record.thread.RunOrdinal, record.thread.Sequence = record.source.Ordinal, event.Sequence + record.thread.CreatedAt, record.thread.Role = event.CreatedAt, "assistant" + records = append(records, record) + } + return records, rows.Err() +} + +func publicThreadReplySource(threadID string, source threadtranscript.Source) (bool, error) { + // Branches copy old assistant messages into new commentary events. They are + // readable history, but their creation is not a new assistant reply. + if source.Event.Source == "thread_branch" || + (source.Event.Type != events.SessionMessageEvent && source.Event.Type != events.ModelPublicCommentaryEvent) { + return false, nil + } + items, err := threadtranscript.Build(threadID, []threadtranscript.Source{source}) + if err != nil { + return false, err + } + for _, item := range items { + if item.Source != "model" || item.Kind != "model_update" || item.Type != threadtranscript.TypeMessage { + continue + } + // Match saved-message search's full-body visibility checks after the Go + // public projection has accepted this source. No text leaves this reader. + var payload struct { + Content string `json:"content"` + Text string `json:"text"` + } + if err := json.Unmarshal([]byte(source.Event.PayloadJSON), &payload); err != nil { + return false, err + } + text := payload.Content + if source.Event.Type == events.ModelPublicCommentaryEvent { + text = payload.Text + } + text = strings.TrimSpace(redact.String(text)) + return text != "" && !syntheticSearchMessage.MatchString(text) && !mockSearchResponse.MatchString(text), nil + } + return false, nil +} + +func encodeThreadReplyCursor(cursor threadReplyCursor) string { + raw, _ := json.Marshal(cursor) + return base64.RawURLEncoding.EncodeToString(raw) +} + +func decodeThreadReplyCursor(encoded, storeID string, head int64) (threadReplyCursor, error) { + invalid := func() (threadReplyCursor, error) { + return threadReplyCursor{}, apperror.New(apperror.CodeInvalidArgument, "reply update cursor does not match this data store or event position") + } + if len(encoded) > 512 { + return invalid() + } + raw, err := base64.RawURLEncoding.DecodeString(encoded) + if err != nil { + return invalid() + } + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.DisallowUnknownFields() + var cursor threadReplyCursor + if err := decoder.Decode(&cursor); err != nil { + return invalid() + } + var trailing any + if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) || cursor.Version != 1 || + cursor.Store != storeID || storeID == "" || cursor.After < 0 || cursor.After > head { + return invalid() + } + return cursor, nil +} diff --git a/internal/store/thread_reply_updates_test.go b/internal/store/thread_reply_updates_test.go new file mode 100644 index 00000000..f8051493 --- /dev/null +++ b/internal/store/thread_reply_updates_test.go @@ -0,0 +1,272 @@ +package store + +import ( + "reflect" + "strings" + "testing" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/application" + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/events" + "cyberagent-workbench/internal/session" +) + +func replyUpdateFixture(t *testing.T) (*SQLiteStore, domain.Run) { + t.Helper() + st := openWorkItemTestStore(t) + _, run, err := application.NewRunService(st).Create(t.Context(), application.CreateRunRequest{ + Goal: "reply updates", Profile: "review", Interactive: true, Budget: domain.Budget{MaxTurns: 8}}) + if err != nil { + t.Fatal(err) + } + return st, run +} + +func replyUpdateBaseline(t *testing.T, st *SQLiteStore) string { + t.Helper() + items, cursor, more, err := st.ListThreadReplyUpdates(t.Context(), "", 100) + if err != nil || len(items) != 0 || cursor == "" || more { + t.Fatalf("baseline: %#v %q %v %v", items, cursor, more, err) + } + return cursor +} + +func appendReplyUpdateEvent(t *testing.T, st *SQLiteStore, run domain.Run, kind, source string, payload any) string { + t.Helper() + event, err := events.New(run.ID, run.MissionID, kind, source, run.ID, payload) + if err != nil { + t.Fatal(err) + } + tx, err := st.db.BeginTx(t.Context(), nil) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback() + event, err = insertRunEventTx(t.Context(), tx, event) + if err != nil { + t.Fatal(err) + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + return event.EventID +} + +func collectReplyUpdates(t *testing.T, st *SQLiteStore, cursor string, limit int) ([]domain.ThreadSearchMatch, string) { + t.Helper() + var all []domain.ThreadSearchMatch + for range 100 { + items, next, more, err := st.ListThreadReplyUpdates(t.Context(), cursor, limit) + if err != nil { + t.Fatal(err) + } + all = append(all, items...) + if !more { + return all, next + } + if cursor == next { + t.Fatal("continuation made no progress") + } + cursor = next + } + t.Fatal("reply update pagination did not finish") + return nil, "" +} + +func replyUpdateIDs(items []domain.ThreadSearchMatch) []string { + ids := make([]string, len(items)) + for i, item := range items { + ids[i] = item.MessageID + } + return ids +} + +func TestThreadReplyUpdatesBaselineInternalPagesAndConcurrentAppend(t *testing.T) { + st, run := replyUpdateFixture(t) + branchReplyEvent(t, st, run, "old answer") + cursor := replyUpdateBaseline(t, st) + appendReplyUpdateEvent(t, st, run, events.ModelStartedEvent, "model", map[string]string{"private": "hidden provider payload"}) + first := branchReplyEvent(t, st, run, "first new answer") + second := branchReplyEvent(t, st, run, "second new answer") + items, next, more, err := st.ListThreadReplyUpdates(t.Context(), cursor, 1) + if err != nil || len(items) != 0 || !more || next == cursor { + t.Fatalf("internal source page: %#v %q %v %v", items, next, more, err) + } + third := branchReplyEvent(t, st, run, "appended after the first page") + got, next := collectReplyUpdates(t, st, next, 1) + if !reflect.DeepEqual(replyUpdateIDs(got), []string{first, second, third}) { + t.Fatalf("append shifted the source cursor: %#v", got) + } + for _, item := range got { + if item.ThreadID != domain.InitialThreadID(run.ID) || item.RunOrdinal != 1 || item.Content != "" || item.ThreadTitle != "" { + t.Fatalf("wrong identity or leaked text: %#v", item) + } + } + items, same, more, err := st.ListThreadReplyUpdates(t.Context(), next, 100) + if err != nil || len(items) != 0 || more || same != next { + t.Fatalf("idle poll: %#v %v", items, err) + } + replayed, _ := collectReplyUpdates(t, st, cursor, 100) + if !reflect.DeepEqual(replyUpdateIDs(replayed), replyUpdateIDs(got)) { + t.Fatal("cursor replay changed identities") + } +} + +func TestThreadReplyUpdatesPublicBoundaryAndDistinctSavedReplyIdentities(t *testing.T) { + st, run := replyUpdateFixture(t) + branchReplyEvent(t, st, run, "same answer") + cursor := replyUpdateBaseline(t, st) + for _, message := range []session.Message{ + session.NewMessage(run.SessionID, "system", "hidden system prompt"), + session.NewEvidenceMessage(run.SessionID, session.SourceToolResult, "private-tool", "private tool output"), + session.NewMessage(run.SessionID, "user", "Continue mission at turn 2 using only the structured tools offered by Go when needed: original request"), + session.NewMessage(run.SessionID, "assistant", "Mock response hidden"), + } { + if _, err := st.SaveSessionMessage(t.Context(), message); err != nil { + t.Fatal(err) + } + } + // Equal text in a later synthetic turn has a new saved identity. The feed + // must not confuse presentation merging with durable operation replay. + repeated := branchReplyEvent(t, st, run, "same answer") + b := branchReplyEvent(t, st, run, "B") + bAgain := branchReplyEvent(t, st, run, "B") + appendReplyUpdateEvent(t, st, run, events.OperatorSteeringQueuedEvent, "operator", map[string]string{"content": "pending or cancelled input"}) + if _, err := st.SaveSessionMessage(t.Context(), session.NewMessage(run.SessionID, "user", "repeat that answer")); err != nil { + t.Fatal(err) + } + afterUser := branchReplyEvent(t, st, run, "same answer") + appendReplyUpdateEvent(t, st, run, events.ToolStartedEvent, "tool", map[string]string{"tool": "read_file", "private": "hidden output"}) + afterTool := branchReplyEvent(t, st, run, "same answer") + long := strings.Repeat("完整答复。", 1500) + longA := branchReplyEvent(t, st, run, long+" A") + longB := branchReplyEvent(t, st, run, long+" B") + commentary := appendReplyUpdateEvent(t, st, run, events.ModelPublicCommentaryEvent, "model", map[string]string{"text": "公开进度说明"}) + appendReplyUpdateEvent(t, st, run, events.ModelPublicCommentaryEvent, "thread_branch", map[string]string{"text": "copied historical answer", "source_event_id": longA}) + got, _ := collectReplyUpdates(t, st, cursor, 1) + if !reflect.DeepEqual(replyUpdateIDs(got), []string{repeated, b, bAgain, afterUser, afterTool, longA, longB, commentary}) { + t.Fatalf("public/duplicate boundary: %#v", replyUpdateIDs(got)) + } +} + +func TestThreadReplyUpdatesLongInternalHistoryUsesOnlySourcePages(t *testing.T) { + st, run := replyUpdateFixture(t) + branchReplyEvent(t, st, run, "already displayed final") + cursor := replyUpdateBaseline(t, st) + for range 250 { + appendReplyUpdateEvent(t, st, run, events.ModelStartedEvent, "model", map[string]string{"private": "internal"}) + } + newReply := branchReplyEvent(t, st, run, "new final") + items, next, more, err := st.ListThreadReplyUpdates(t.Context(), cursor, 100) + if err != nil || len(items) != 0 || !more || next == cursor { + t.Fatalf("bounded source page: %#v %v %v", items, more, err) + } + got, _ := collectReplyUpdates(t, st, next, 100) + if !reflect.DeepEqual(replyUpdateIDs(got), []string{newReply}) { + t.Fatalf("internal pages skipped or duplicated a reply: %#v", got) + } +} + +func TestThreadReplyUpdatesCrossRunArchiveDeleteAndRealBranch(t *testing.T) { + st, run := replyUpdateFixture(t) + selected := branchReplyEvent(t, st, run, "source answer") + cursor := replyUpdateBaseline(t, st) + branch, err := application.NewThreadBranchService(st).Branch(t.Context(), domain.ThreadBranchRequest{ + Version: domain.ThreadBranchProtocolVersion, ThreadID: domain.InitialThreadID(run.ID), + SourceEventID: selected, OperationKey: "reply-update-branch", RequestedBy: "test_operator"}) + if err != nil { + t.Fatal(err) + } + if copied, _ := collectReplyUpdates(t, st, cursor, 100); len(copied) != 0 { + t.Fatalf("branch copy was a reply: %#v", copied) + } + branchNew := branchReplyEvent(t, st, branch.Run, "new answer in branch") + runs := application.NewRunService(st) + if _, err := runs.Start(t.Context(), run.ID); err != nil { + t.Fatal(err) + } + if _, err := runs.Fail(t.Context(), run.ID, "continue fixture"); err != nil { + t.Fatal(err) + } + continued, err := application.NewThreadService(st).Submit(t.Context(), application.SubmitThreadMessageRequest{ + Version: domain.ThreadMessageProtocolVersion, ThreadID: domain.InitialThreadID(run.ID), Content: "next run", + OperationKey: "reply-updates-successor", RequestedBy: "test_operator"}) + if err != nil { + t.Fatal(err) + } + successor := branchReplyEvent(t, st, continued.Run, "source answer") + if _, err := runs.Start(t.Context(), continued.Run.ID); err != nil { + t.Fatal(err) + } + if _, err := runs.Fail(t.Context(), continued.Run.ID, "archive fixture"); err != nil { + t.Fatal(err) + } + thread, err := st.GetThread(t.Context(), continued.Thread.ID) + if err != nil { + t.Fatal(err) + } + archived, err := application.NewThreadService(st).Transition(t.Context(), thread.ID, domain.ThreadArchive, thread.Version, "test_operator", "reply-updates-archive") + if err != nil { + t.Fatal(err) + } + got, _ := collectReplyUpdates(t, st, cursor, 100) + if !reflect.DeepEqual(replyUpdateIDs(got), []string{branchNew, successor}) || got[1].RunOrdinal != 2 || got[1].ThreadStatus != domain.ThreadArchived { + t.Fatalf("cross-Run/archive updates: %#v", got) + } + if _, err := application.NewThreadService(st).Transition(t.Context(), thread.ID, domain.ThreadDelete, archived.Version, "test_operator", "reply-updates-delete"); err != nil { + t.Fatal(err) + } + got, _ = collectReplyUpdates(t, st, cursor, 100) + if !reflect.DeepEqual(replyUpdateIDs(got), []string{branchNew}) { + t.Fatalf("deleted reply exposed: %#v", got) + } +} + +func TestThreadReplyUpdatesErrorsNeverReturnPartialCursorAndRejectOtherStores(t *testing.T) { + st, run := replyUpdateFixture(t) + cursor := replyUpdateBaseline(t, st) + first := branchReplyEvent(t, st, run, "good answer") + second := branchReplyEvent(t, st, run, "answer with broken source") + var created string + if err := st.db.QueryRow(`SELECT created_at FROM run_events WHERE event_id=?`, second).Scan(&created); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(`UPDATE run_events SET created_at='' WHERE event_id=?`, second); err != nil { + t.Fatal(err) + } + items, next, more, err := st.ListThreadReplyUpdates(t.Context(), cursor, 100) + if err == nil || items != nil || next != "" || more { + t.Fatalf("partial success escaped: %#v %q %v %v", items, next, more, err) + } + if _, err := st.db.Exec(`UPDATE run_events SET created_at=? WHERE event_id=?`, created, second); err != nil { + t.Fatal(err) + } + got, _ := collectReplyUpdates(t, st, cursor, 100) + if !reflect.DeepEqual(replyUpdateIDs(got), []string{first, second}) { + t.Fatal("retry lost the first reply") + } + other, _ := replyUpdateFixture(t) + for _, bad := range []string{replyUpdateBaseline(t, other), "not-a-cursor", encodeThreadReplyCursor(threadReplyCursor{Version: 1, Store: st.DataStoreID(), After: 1 << 60})} { + if _, _, _, err := st.ListThreadReplyUpdates(t.Context(), bad, 100); apperror.CodeOf(err) != apperror.CodeInvalidArgument { + t.Fatalf("invalid cursor accepted: %v", err) + } + } +} + +func TestThreadReplyUpdatesMetadataChangesDoNotCreateReplies(t *testing.T) { + st, run := replyUpdateFixture(t) + branchReplyEvent(t, st, run, "existing reply") + cursor := replyUpdateBaseline(t, st) + threadID := domain.InitialThreadID(run.ID) + if _, err := st.RenameThread(t.Context(), threadID, "renamed chat"); err != nil { + t.Fatal(err) + } + if _, err := st.PinThread(t.Context(), threadID, true); err != nil { + t.Fatal(err) + } + items, next, more, err := st.ListThreadReplyUpdates(t.Context(), cursor, 100) + if err != nil || len(items) != 0 || more || next != cursor { + t.Fatalf("metadata looked like a new reply: %#v %v", items, err) + } +} diff --git a/protocols/registry.json b/protocols/registry.json index 75c32018..6dad1895 100644 --- a/protocols/registry.json +++ b/protocols/registry.json @@ -82,7 +82,8 @@ "operator_action_item.v1", "run-event-poll.v1", "run-events.v1", - "run_activity.v1" + "run_activity.v1", + "thread_reply_updates.v1" ], "retired_identifiers": [] }, diff --git a/web/src/api/schema.d.ts b/web/src/api/schema.d.ts index 0d48cb4b..edb57dfd 100644 --- a/web/src/api/schema.d.ts +++ b/web/src/api/schema.d.ts @@ -4712,6 +4712,26 @@ export interface paths { patch?: never; trace?: never; }; + "/api/v1/threads/reply-updates": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * Observe new saved public replies across conversations + * @description Read-only incremental observation of saved public assistant message identities in active and archived conversations. Omitting cursor establishes the current baseline with no historical items. Subsequent requests scan at most limit source events in global save order; an empty items array can still have has_more=true. Continue using data.cursor until has_more=false, then retain that cursor for the next poll. Cursor scope includes the exact data_store_id. Persist returned message identities before the cursor; an error returns no partial cursor. Replaying a cursor preserves message identities. Separate saved messages with equal text remain separate updates. Excludes deleted conversations, copied branch history, users, tools, hidden/internal events and provisional streaming bytes. No message body or private event payload is returned. + */ + get: operations["listThreadReplyUpdates"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/v1/threads/search": { parameters: { query?: never; @@ -15235,6 +15255,26 @@ export interface components { /** @enum {string} */ version: "thread_queued_messages.v1"; }; + ThreadReplyUpdateView: { + /** Format: date-time */ + created_at: string; + message_id: string; + /** Format: int64 */ + run_ordinal: number; + /** Format: int64 */ + sequence: number; + thread_id: string; + /** @enum {string} */ + thread_status: "active" | "archived"; + }; + ThreadReplyUpdatesView: { + cursor: string; + data_store_id: string; + has_more: boolean; + items: components["schemas"]["ThreadReplyUpdateView"][]; + /** @enum {string} */ + version: "thread_reply_updates.v1"; + }; ThreadRequestObservationView: { error_code?: string; failure_stage?: string; @@ -28261,6 +28301,44 @@ export interface operations { 504: components["responses"]["GatewayTimeout"]; }; }; + listThreadReplyUpdates: { + parameters: { + query?: { + /** @description Maximum source events scanned, including filtered events; not a requested number of visible replies */ + limit?: number; + /** @description Opaque cursor from data.cursor, bound to the exact data store; omit only to establish a new baseline */ + cursor?: string; + }; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Successful read */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + data: components["schemas"]["ThreadReplyUpdatesView"]; + request_id: string; + /** @constant */ + version: "api.v1"; + }; + }; + }; + 400: components["responses"]["BadRequest"]; + 401: components["responses"]["Unauthorized"]; + 403: components["responses"]["Forbidden"]; + 414: components["responses"]["RequestTooLarge"]; + 429: components["responses"]["ResourceExhausted"]; + 500: components["responses"]["InternalError"]; + 503: components["responses"]["Unavailable"]; + 504: components["responses"]["GatewayTimeout"]; + }; + }; searchSavedThreadMessages: { parameters: { query: { diff --git a/web/src/api/types.ts b/web/src/api/types.ts index cedf49f5..3ffde521 100644 --- a/web/src/api/types.ts +++ b/web/src/api/types.ts @@ -115,6 +115,8 @@ export type ThreadMessageView = components["schemas"]["ThreadMessageView"]; export type ThreadTranscriptItemView = components["schemas"]["ThreadTranscriptItemView"]; export type ThreadSearchResultView = components["schemas"]["ThreadSearchResultView"]; export type ThreadSavedMessageView = components["schemas"]["ThreadSavedMessageView"]; +export type ThreadReplyUpdateView = components["schemas"]["ThreadReplyUpdateView"]; +export type ThreadReplyUpdatesView = components["schemas"]["ThreadReplyUpdatesView"]; export type ThreadActivityTypedDetailView = components["schemas"]["ThreadActivityTypedDetailView"]; export type ThreadActivityKind = ThreadActivityTypedDetailView["kind"]; diff --git a/web/src/v2/app.test.tsx b/web/src/v2/app.test.tsx index 3dd45bc1..4efe9931 100644 --- a/web/src/v2/app.test.tsx +++ b/web/src/v2/app.test.tsx @@ -155,6 +155,33 @@ it.each([false, true])("moves the text draft while retaining a late file referen }); describe("V2Workbench inspector navigation", () => { + it("keeps a restored manual unread marker until an explicit sidebar click, including the current chat", async () => { + const scope = `ds1_${"9".repeat(64)}`; + useConnectionStore.getState().setHealth({ status: "ok", api_version: "api.v1", app_version: "fixture", schema_version: 216, data_store_id: scope }); + const key = `v2_recovery.v1:${encodeURIComponent(window.location.origin)}:${encodeURIComponent("/api/v1")}:${scope}:reply-read-state`; + localStorage.setItem(key, JSON.stringify({ version: "v2_recovery.v1", value: { cursor: "known-feed", threads: { + [createdThread.id]: { pending: [], seen: [], manualUnread: true }, + } } })); + window.history.replaceState({}, "", `#/threads/${createdThread.id}`); + const client = Object.assign(navigationClient(), { baseURL: "/api/v1" }); + vi.mocked(client.get).mockResolvedValue({ version: "thread_reply_updates.v1", data_store_id: scope, cursor: "known-feed", has_more: false, items: [] }); + render( + ); + const row = await screen.findByRole("button", { name: createdThread.title }); + await waitFor(() => expect(client.get).toHaveBeenCalledWith("/threads/reply-updates", { limit: 100, cursor: "known-feed" }, expect.any(AbortSignal))); + expect(row).toHaveAccessibleDescription(expect.stringContaining("未读")); + const user = userEvent.setup(); + await user.click(row); + expect(row).not.toHaveAccessibleDescription(expect.stringContaining("未读")); + expect(window.location.hash).toBe(`#/threads/${createdThread.id}`); + await user.click(screen.getByRole("button", { name: "First turn 聊天选项" })); + await user.click(screen.getByRole("menuitem", { name: "标为未读" })); + expect(row).toHaveAccessibleDescription(expect.stringContaining("未读")); + await user.click(row); + expect(row).not.toHaveAccessibleDescription(expect.stringContaining("未读")); + expectNoNavigationWrites(client); + }); + it("offers the new branch without replacing a chat selected while branching", async () => { const otherThread = { ...createdThread, id: "thread-other", title: "Other conversation" }; const branched = { ...createdThread, id: "thread-branch", title: "Branched conversation" }; diff --git a/web/src/v2/app.tsx b/web/src/v2/app.tsx index 1ff656a4..7d1f473f 100644 --- a/web/src/v2/app.tsx +++ b/web/src/v2/app.tsx @@ -16,6 +16,7 @@ import { V2SidebarLayout } from "./components/sidebar-layout"; import { V2Titlebar } from "./components/titlebar"; import { useV2Client } from "./client-session"; import { v2QueryKeys } from "./query-keys"; +import { useV2ThreadUnread } from "./use-thread-unread"; import { registerV2RecoveredTurn, useV2RestoreTurns, useV2ThreadTurn, v2TurnFailed } from "./use-thread-turn"; import { ProjectDialog, V2WorkspaceStart } from "./components/workspace-start"; import { useV2Navigation, type V2RunPane } from "./navigation"; @@ -258,6 +259,9 @@ function V2WorkbenchContent({ client }: { client: APIClient }) { useV2RestoreTurns(); const recovery = useV2RecoveryStore(); const persistenceWarning = useV2PersistenceWarning(); + const scopeID = useConnectionStore((state) => state.health?.data_store_id); + const unread = useV2ThreadUnread(client, scopeID); + const [readRequest, setReadRequest] = useState({ threadID: "", revision: 0 }); useEffect(() => { document.documentElement.dataset.prayuDensity = readDensity(); }, []); @@ -537,10 +541,10 @@ function V2WorkbenchContent({ client }: { client: APIClient }) { const closeNavigationSidebar = () => { if (window.matchMedia?.("(max-width: 720px)").matches) setSidebarVisible(false); }; - const openConversation = (threadID: string) => { + const openConversation = (threadID: string, onAccepted?: () => void) => { setBackgroundCreations((current) => current.filter((thread) => thread.id !== threadID)); updateModelSetupToken(""); - navigate({ kind: "thread", threadID, ...(route.view ? { view: route.view } : {}) }); + navigate({ kind: "thread", threadID, ...(route.view ? { view: route.view } : {}) }, false, onAccepted); closeNavigationSidebar(); }; const startNew = (destination?: string) => { @@ -657,13 +661,18 @@ function V2WorkbenchContent({ client }: { client: APIClient }) { setSearchOpen(open); setSearchCurrentThread(false); if (!open) { setThreadSearch(""); setSearchComposing(false); } - }} onSelectThread={openConversation} searchOpen={searchOpen} + }} onSelectThread={(id) => openConversation(id, () => { + unread.open(id); + setReadRequest((current) => ({ threadID: id, revision: current.revision + 1 })); + })} searchOpen={searchOpen} + unreadThreadIDs={unread.unreadThreadIDs} onReadStateChange={unread.setRead} search={threadSearch} appliedSearch={appliedThreadSearch} searchPending={searchPending} searchTooLong={searchTooLong} searchResults={searchResults} includeArchived={includeArchivedSearch} onIncludeArchivedChange={setIncludeArchivedSearch} searchCurrentThread={Boolean(searchThreadID)} onSearchCurrentThreadChange={setSearchCurrentThread} onSelectSearchResult={(result) => { - navigate({ kind: "thread", threadID: result.thread_id, ...(result.message_id ? { messageID: result.message_id } : {}) }); + navigate({ kind: "thread", threadID: result.thread_id, ...(result.message_id ? { messageID: result.message_id } : {}) }, + false, () => unread.open(result.thread_id)); closeNavigationSidebar(); }} onSearchChange={(value) => { setThreadSearch(value); if (!value) setSearchComposing(false); }} @@ -790,6 +799,8 @@ function V2WorkbenchContent({ client }: { client: APIClient }) { type="button">{t("加载更多项目", "Load more projects")}} workspaceID={workspaceID} workspaces={workspaces} /> : { if (routeRef.current === route) openConversation(thread.id); @@ -826,6 +837,10 @@ function V2WorkbenchContent({ client }: { client: APIClient }) { } {persistenceWarning &&
{persistenceWarning}
} + {unread.error &&
+ {t("未读回复暂未同步，已有标记会保留。", "Unread replies could not be refreshed. Existing markers have been kept.")} + +
}