kurihada/ai-workflow-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement inbox read cursors for unread threads

Browse Source
This commit is contained in:
kurihada
2026-03-19 03:43:10 +08:00
parent 02d98a78dd
commit 1927930570
19 changed files with 240 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cli/inbox/cancel.go
+1 -2
View File
@@ -3,7 +3,6 @@ package inbox
import (
"fmt"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -38,7 +37,7 @@ func newCancelCmd(root *rootOptions) *cobra.Command {
return err
}
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
internal/cli/inbox/claim.go
+1 -2
View File
@@ -4,7 +4,6 @@ import (
"errors"
"fmt"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -34,7 +33,7 @@ func newClaimCmd(root *rootOptions) *cobra.Command {
return protocol.InvalidInput("agent is required", nil)
}
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
internal/cli/inbox/db.go
+22
View File
@@ -0,0 +1,22 @@
package inbox
import (
"context"
"database/sql"
"ai-workflow-skill/internal/db"
)
func openInboxDB(ctx context.Context, dbPath string) (*sql.DB, error) {
sqlDB, err := db.Open(ctx, dbPath)
if err != nil {
return nil, err
}
if err := db.ApplyMigrations(ctx, sqlDB); err != nil {
_ = sqlDB.Close()
return nil, err
}
return sqlDB, nil
}
internal/cli/inbox/done.go
+1 -2
View File
@@ -3,7 +3,6 @@ package inbox
import (
"fmt"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -54,7 +53,7 @@ func newCompleteCmd(root *rootOptions, mode string) *cobra.Command {
return err
}
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
internal/cli/inbox/fetch.go
+1 -2
View File
@@ -4,7 +4,6 @@ import (
"fmt"
"strings"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -32,7 +31,7 @@ func newFetchCmd(root *rootOptions) *cobra.Command {
agent = root.agent
}
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
internal/cli/inbox/init.go
+1 -6
View File
@@ -3,7 +3,6 @@ package inbox
import (
"fmt"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"github.com/spf13/cobra"
@@ -16,16 +15,12 @@ func newInitCmd(opts *rootOptions) *cobra.Command {
RunE: func(cmd *cobra.Command, args []string) error {
ctx := cmd.Context()
sqlDB, err := db.Open(ctx, opts.dbPath)
sqlDB, err := openInboxDB(ctx, opts.dbPath)
if err != nil {
return err
}
defer sqlDB.Close()
if err := db.ApplyMigrations(ctx, sqlDB); err != nil {
return err
}
resp := protocol.Success{
OK: true,
Command: "init",
internal/cli/inbox/integration_test.go
+103
View File
@@ -316,6 +316,7 @@ func TestInboxRenewWaitReplyAndCancel(t *testing.T) {
go func() {
stdout, stderr, exitCode := executeInboxCommand(
"--db", dbPath,
"--agent", "worker-c",
"--json",
"wait-reply",
"--thread", threadID,
@@ -359,6 +360,18 @@ func TestInboxRenewWaitReplyAndCancel(t *testing.T) {
t.Fatalf("expected answer wake message, got %q", kind)
}
stdout, _, exitCode := executeInboxCommand(
"--db", dbPath,
"--agent", "worker-c",
"--json",
"fetch",
"--status", "blocked",
"--unread",
)
if exitCode != 10 {
t.Fatalf("expected blocked unread list to be cleared after wait-reply, got exit %d with %s", exitCode, stdout)
}
cancelOut := runInboxCommand(
t,
"--db", dbPath,
@@ -533,6 +546,96 @@ func TestInboxWatchListUnreadAndAppend(t *testing.T) {
}
}
func TestInboxUnreadReadCursor(t *testing.T) {
t.Parallel()
dbPath := filepath.Join(t.TempDir(), "coord.db")
runInboxCommand(t, "--db", dbPath, "--json", "init")
sendOut := runInboxCommand(
t,
"--db", dbPath,
"--json",
"send",
"--from", "leader",
"--to", "worker-e",
"--subject", "Review navbar copy",
"--summary", "Check top nav wording",
)
var sendResp map[string]any
mustDecodeJSON(t, sendOut, &sendResp)
threadID := nestedString(t, sendResp, "data", "thread", "thread_id")
fetchOut := runInboxCommand(
t,
"--db", dbPath,
"--json",
"fetch",
"--agent", "worker-e",
"--status", "pending",
"--unread",
)
var fetchResp map[string]any
mustDecodeJSON(t, fetchOut, &fetchResp)
threads, ok := nestedValue(t, fetchResp, "data", "threads").([]any)
if !ok || len(threads) != 1 {
t.Fatalf("expected one unread pending thread, got %#v", nestedValue(t, fetchResp, "data", "threads"))
}
runInboxCommand(
t,
"--db", dbPath,
"--agent", "worker-e",
"--json",
"show",
"--thread", threadID,
"--mark-read",
)
stdout, _, exitCode := executeInboxCommand(
"--db", dbPath,
"--json",
"fetch",
"--agent", "worker-e",
"--status", "pending",
"--unread",
)
if exitCode != 10 {
t.Fatalf("expected unread fetch to clear after mark-read, got exit %d with %s", exitCode, stdout)
}
runInboxCommand(
t,
"--db", dbPath,
"--json",
"send",
"--from", "leader",
"--to", "worker-e",
"--thread", threadID,
"--summary", "Use sentence case",
"--body", "Keep the nav labels in sentence case.",
)
fetchOut = runInboxCommand(
t,
"--db", dbPath,
"--json",
"fetch",
"--agent", "worker-e",
"--status", "pending",
"--unread",
)
mustDecodeJSON(t, fetchOut, &fetchResp)
threads, ok = nestedValue(t, fetchResp, "data", "threads").([]any)
if !ok || len(threads) != 1 {
t.Fatalf("expected unread thread to reappear after new message, got %#v", nestedValue(t, fetchResp, "data", "threads"))
}
}
func TestInboxJSONErrorsAndExitCodes(t *testing.T) {
t.Parallel()
internal/cli/inbox/list.go
+1 -2
View File
@@ -3,7 +3,6 @@ package inbox
import (
"fmt"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -32,7 +31,7 @@ func newListCmd(root *rootOptions) *cobra.Command {
agent = root.agent
}
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
internal/cli/inbox/renew.go
+1 -2
View File
@@ -3,7 +3,6 @@ package inbox
import (
"fmt"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -33,7 +32,7 @@ func newRenewCmd(root *rootOptions) *cobra.Command {
return protocol.InvalidInput("agent is required", nil)
}
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
internal/cli/inbox/reply.go
+1 -2
View File
@@ -3,7 +3,6 @@ package inbox
import (
"fmt"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -48,7 +47,7 @@ func newReplyCmd(root *rootOptions) *cobra.Command {
return err
}
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
internal/cli/inbox/send.go
+1 -2
View File
@@ -3,7 +3,6 @@ package inbox
import (
"fmt"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -55,7 +54,7 @@ func newSendCmd(root *rootOptions) *cobra.Command {
return err
}
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
internal/cli/inbox/show.go
+9 -3
View File
@@ -3,7 +3,6 @@ package inbox
import (
"fmt"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -12,6 +11,7 @@ import (
type showOptions struct {
threadID string
markRead bool
}
func newShowCmd(root *rootOptions) *cobra.Command {
@@ -23,14 +23,19 @@ func newShowCmd(root *rootOptions) *cobra.Command {
RunE: func(cmd *cobra.Command, args []string) error {
ctx := cmd.Context()
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
defer sqlDB.Close()
s := store.NewInboxStore(sqlDB)
detail, err := s.GetThread(ctx, opts.threadID)
agent := root.agent
if opts.markRead && agent == "" {
return protocol.InvalidInput("agent is required when using --mark-read", nil)
}
detail, err := s.GetThreadForAgent(ctx, opts.threadID, agent, opts.markRead)
if err != nil {
return err
}
@@ -66,6 +71,7 @@ func newShowCmd(root *rootOptions) *cobra.Command {
}
cmd.Flags().StringVar(&opts.threadID, "thread", "", "Thread ID")
cmd.Flags().BoolVar(&opts.markRead, "mark-read", false, "Advance the caller's read cursor to the latest message")
_ = cmd.MarkFlagRequired("thread")
return cmd
internal/cli/inbox/update.go
+1 -2
View File
@@ -3,7 +3,6 @@ package inbox
import (
"fmt"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -47,7 +46,7 @@ func newUpdateCmd(root *rootOptions) *cobra.Command {
return err
}
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
internal/cli/inbox/wait_reply.go
+3 -2
View File
@@ -4,7 +4,6 @@ import (
"fmt"
"time"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -28,18 +27,20 @@ func newWaitReplyCmd(root *rootOptions) *cobra.Command {
RunE: func(cmd *cobra.Command, args []string) error {
ctx := cmd.Context()
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
defer sqlDB.Close()
s := store.NewInboxStore(sqlDB)
agent := root.agent
result, err := s.WaitReply(ctx, store.WaitReplyInput{
ThreadID: opts.threadID,
AfterMessageID: opts.afterMessageID,
AfterEventID: opts.afterEventID,
Kinds: parseCSV(opts.kinds),
Agent: agent,
Timeout: time.Duration(opts.timeoutSeconds) * time.Second,
})
if err != nil {
internal/cli/inbox/watch.go
+1 -2
View File
@@ -4,7 +4,6 @@ import (
"fmt"
"time"
"ai-workflow-skill/internal/db"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
@@ -32,7 +31,7 @@ func newWatchCmd(root *rootOptions) *cobra.Command {
agent = root.agent
}
sqlDB, err := db.Open(ctx, root.dbPath)
sqlDB, err := openInboxDB(ctx, root.dbPath)
if err != nil {
return err
}
internal/db/schema/005_inbox_reads.sql
+12
View File
@@ -0,0 +1,12 @@
CREATE TABLE IF NOT EXISTS thread_reads (
thread_id TEXT NOT NULL,
agent_id TEXT NOT NULL,
last_read_message_id TEXT NOT NULL,
last_read_at TEXT NOT NULL,
PRIMARY KEY(thread_id, agent_id),
FOREIGN KEY(thread_id) REFERENCES threads(thread_id),
FOREIGN KEY(last_read_message_id) REFERENCES messages(message_id)
);
CREATE INDEX IF NOT EXISTS idx_thread_reads_agent
ON thread_reads(agent_id, last_read_at);
internal/store/inbox.go
+62 -6
View File
@@ -191,6 +191,7 @@ type WaitReplyInput struct {
AfterMessageID string
AfterEventID int64
Kinds []string
Agent string
Timeout time.Duration
}
@@ -412,7 +413,8 @@ func (s *InboxStore) ListThreads(ctx context.Context, input ListInput) ([]Thread
}
var (
args []any
joinArgs []any
whereArgs []any
conditions []string
joins []string
)
@@ -424,16 +426,16 @@ func (s *InboxStore) ListThreads(ctx context.Context, input ListInput) ([]Thread
if assignedTo != "" {
conditions = append(conditions, "t.assigned_to = ?")
args = append(args, assignedTo)
whereArgs = append(whereArgs, assignedTo)
}
if input.CreatedBy != "" {
conditions = append(conditions, "t.created_by = ?")
args = append(args, input.CreatedBy)
whereArgs = append(whereArgs, input.CreatedBy)
}
if len(input.Statuses) > 0 {
conditions = append(conditions, "t.status IN ("+placeholders(len(input.Statuses))+")")
for _, status := range input.Statuses {
args = append(args, status)
whereArgs = append(whereArgs, status)
}
}
if input.Unread {
@@ -441,10 +443,13 @@ func (s *InboxStore) ListThreads(ctx context.Context, input ListInput) ([]Thread
return nil, fmt.Errorf("%w: agent is required when filtering unread threads", ErrInvalidInput)
}
joins = append(joins, "JOIN messages lm ON lm.message_id = t.latest_message_id")
joins = append(joins, "LEFT JOIN thread_reads tr ON tr.thread_id = t.thread_id AND tr.agent_id = ?")
joinArgs = append(joinArgs, input.Agent)
conditions = append(conditions, "lm.to_agent = ?")
args = append(args, input.Agent)
whereArgs = append(whereArgs, input.Agent)
conditions = append(conditions, "lm.from_agent <> ?")
args = append(args, input.Agent)
whereArgs = append(whereArgs, input.Agent)
conditions = append(conditions, "(tr.last_read_message_id IS NULL OR tr.last_read_message_id <> t.latest_message_id)")
}
query := `SELECT
@@ -458,6 +463,7 @@ func (s *InboxStore) ListThreads(ctx context.Context, input ListInput) ([]Thread
query += " WHERE " + strings.Join(conditions, " AND ")
}
query += " ORDER BY t.updated_at DESC LIMIT ?"
args := append(joinArgs, whereArgs...)
args = append(args, limit)
rows, err := s.db.QueryContext(ctx, query, args...)
@@ -1078,6 +1084,10 @@ func (s *InboxStore) CancelThread(ctx context.Context, input CancelInput) (Threa
}
func (s *InboxStore) GetThread(ctx context.Context, threadID string) (ThreadDetail, error) {
return s.GetThreadForAgent(ctx, threadID, "", false)
}
func (s *InboxStore) GetThreadForAgent(ctx context.Context, threadID, agent string, markRead bool) (ThreadDetail, error) {
thread, err := selectThread(ctx, s.db, threadID)
if err != nil {
return ThreadDetail{}, err
@@ -1117,6 +1127,12 @@ func (s *InboxStore) GetThread(ctx context.Context, threadID string) (ThreadDeta
}
attachArtifacts(messages, artifactsByMessageID)
if markRead {
if err := markThreadRead(ctx, s.db, thread.ThreadID, agent, thread.LatestMessageID, nowUTC()); err != nil {
return ThreadDetail{}, err
}
}
return ThreadDetail{
Thread: thread,
Messages: messages,
@@ -1204,6 +1220,11 @@ func (s *InboxStore) WaitReply(ctx context.Context, input WaitReplyInput) (WaitR
return WaitReplyResult{}, err
}
if found {
if shouldMarkMessageRead(message, input.Agent) {
if err := markThreadRead(waitCtx, s.db, input.ThreadID, input.Agent, message.MessageID, nowUTC()); err != nil {
return WaitReplyResult{}, err
}
}
return WaitReplyResult{
Woke: true,
NextEventID: eventID,
@@ -1363,6 +1384,10 @@ type queryRower interface {
QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}
type execContexter interface {
ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
}
type eventInput struct {
RunID string
TaskID string
@@ -1482,6 +1507,30 @@ func updateThreadState(ctx context.Context, tx *sql.Tx, threadID, status, assign
return nil
}
func markThreadRead(ctx context.Context, execer execContexter, threadID, agent, messageID string, readAt time.Time) error {
if agent == "" || messageID == "" {
return nil
}
_, err := execer.ExecContext(
ctx,
`INSERT INTO thread_reads (
thread_id, agent_id, last_read_message_id, last_read_at
) VALUES (?, ?, ?, ?)
ON CONFLICT(thread_id, agent_id) DO UPDATE SET
last_read_message_id = excluded.last_read_message_id,
last_read_at = excluded.last_read_at`,
threadID,
agent,
messageID,
formatTime(readAt),
)
if err != nil {
return fmt.Errorf("mark thread read: %w", err)
}
return nil
}
func loadArtifactsForMessageIDs(ctx context.Context, db *sql.DB, messageIDs []string) (map[string][]Artifact, error) {
result := make(map[string][]Artifact)
if len(messageIDs) == 0 {
@@ -1809,6 +1858,13 @@ func isSQLiteBusyError(err error) bool {
strings.Contains(message, "database table is locked")
}
func shouldMarkMessageRead(message Message, agent string) bool {
if agent == "" {
return false
}
return message.ToAgent == agent && message.FromAgent != agent
}
func defaultID(value, prefix string) string {
if value != "" {
return value