fix(chat-uploads): align poll-mode activity rows with inbox poll filter

The workspace inbox poller filters
`GET /workspaces/:id/activity?type=a2a_receive` — writing rows with
`activity_type=chat_upload_receive` would be silently invisible to it.

Switch the poll-mode upload-staging handler to write
`activity_type=a2a_receive` with `method=chat_upload_receive` as the
discriminator. Same shape as A2A's `tasks/send` vs `message/send` method
split; the workspace-side handler (Phase 2) routes by `method`, not
activity_type.

Pinned with `TestPollUpload_ActivityRowDiscriminator` — sqlmock
WithArgs on positions 2 (activity_type) and 5 (method) so a refactor
that flips activity_type back to a custom value gets a red test
instead of a runtime "poller saw nothing" silent break.
This commit is contained in:
Hongming Wang 2026-05-05 04:29:07 -07:00
parent 86fdaad111
commit 43bf94a07c
2 changed files with 77 additions and 7 deletions

View File

@ -651,18 +651,19 @@ func (h *ChatFilesHandler) uploadPollMode(c *gin.Context, ctx context.Context, w
}
// Activity row so the workspace's inbox poller picks this up
// on its next cycle. type=chat_upload_receive is a new
// activity_type the workspace's message_from_activity adapter
// (Phase 2) will handle by fetching content via the GET
// endpoint. The request_body carries everything the
// workspace needs to dispatch the fetch — including the
// synthetic URI canvas embeds in the chat message.
// on its next cycle. activity_type=a2a_receive (NOT a new
// type) so the existing poll filter
// `?type=a2a_receive` catches it without poll-side changes;
// method=chat_upload_receive is the discriminator the
// workspace's adapter (Phase 2) uses to route to the upload
// fetcher instead of the agent's message handler. Same
// shape as A2A's tasks/send vs message/send method split.
uri := fmt.Sprintf("platform-pending:%s/%s", workspaceID, fileID)
summary := "chat_upload_receive: " + sanitized
method := "chat_upload_receive"
LogActivity(ctx, h.broadcaster, ActivityParams{
WorkspaceID: workspaceID,
ActivityType: "chat_upload_receive",
ActivityType: "a2a_receive",
TargetID: &workspaceID,
Method: &method,
Summary: &summary,

View File

@ -96,6 +96,37 @@ func expectActivityInsert(mock sqlmock.Sqlmock) {
WillReturnResult(sqlmock.NewResult(1, 1))
}
// expectActivityInsertWithTypeAndMethod is a strict variant that pins
// the activity_type and method positional args. Used in the discriminator
// regression test below — the workspace inbox poller filters
// `?type=a2a_receive`, so writing any other activity_type silently breaks
// poll-mode delivery without a build/test error. Pin the two discriminator
// fields so a refactor that flips activity_type back to a custom value is
// caught here instead of at runtime by a confused poller.
//
// Positional args (LogActivity uses ExecContext with 12 positional params):
// $1 workspace_id, $2 activity_type, $3 source_id, $4 target_id,
// $5 method, $6 summary, $7 request_body, $8 response_body,
// $9 tool_trace, $10 duration_ms, $11 status, $12 error_detail.
func expectActivityInsertWithTypeAndMethod(mock sqlmock.Sqlmock, workspaceID, activityType, method string) {
mock.ExpectExec(`INSERT INTO activity_logs`).
WithArgs(
workspaceID, // $1 workspace_id
activityType, // $2 activity_type ← pinned
sqlmock.AnyArg(), // $3 source_id
sqlmock.AnyArg(), // $4 target_id (workspaceID, but already covered)
method, // $5 method ← pinned
sqlmock.AnyArg(), // $6 summary
sqlmock.AnyArg(), // $7 request_body
sqlmock.AnyArg(), // $8 response_body
sqlmock.AnyArg(), // $9 tool_trace
sqlmock.AnyArg(), // $10 duration_ms
sqlmock.AnyArg(), // $11 status
sqlmock.AnyArg(), // $12 error_detail
).
WillReturnResult(sqlmock.NewResult(1, 1))
}
// pollUploadFixture builds a multipart body with N named files.
func pollUploadFixture(t *testing.T, files map[string][]byte) (*bytes.Buffer, string) {
t.Helper()
@ -518,3 +549,41 @@ func TestPollUpload_SanitizesFilenameInResponse(t *testing.T) {
t.Errorf("storage Put didn't receive sanitized filename: %+v", store.puts)
}
}
// TestPollUpload_ActivityRowDiscriminator pins the
// activity_type / method shape that the workspace inbox poller depends
// on. The poller filters `GET /workspaces/:id/activity?type=a2a_receive`
// so the handler MUST write activity_type=a2a_receive (NOT a custom
// type), and use method=chat_upload_receive as the
// upload-vs-message-vs-task discriminator.
//
// Why pinned: a previous iteration of this handler used
// activity_type="chat_upload_receive" — silently invisible to the
// existing poller. The branch passed every push-mode test, every
// storage test, and every per-file content test; the bug only
// surfaced at runtime when the workspace polled and got nothing.
// Encode the contract in a unit test so the next refactor can't
// re-break it without a red CI.
func TestPollUpload_ActivityRowDiscriminator(t *testing.T) {
mock := setupTestDB(t)
setupTestRedis(t)
wsID := "abc12345-6789-4abc-8def-000000000999"
expectPollDeliveryMode(mock, wsID, "poll")
expectActivityInsertWithTypeAndMethod(mock, wsID, "a2a_receive", "chat_upload_receive")
store := newInMemStorage()
h := NewChatFilesHandler(NewTemplatesHandler(t.TempDir(), nil)).
WithPendingUploads(store, nil)
body, ct := pollUploadFixture(t, map[string][]byte{"x.pdf": []byte("xx")})
c, w := makeUploadRequest(t, wsID, body, ct)
h.Upload(c)
if w.Code != http.StatusOK {
t.Fatalf("status=%d body=%s", w.Code, w.Body.String())
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("expectations: %v", err)
}
}