feat(taskrun): finalize RunManifest in one unit of work
CAS the unique terminal status with the assistant turn, frozen manifest, envelope and FAV staging, and join ctxmgr writes through the same UoW instead of a process mutex. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -5,6 +5,11 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"chat-agent-system/internal/app/uow"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// UpdateMessageMetadata 更新已持久化助手消息的 metadata 与任务状态。
|
||||
@@ -19,9 +24,9 @@ func (m *ContextManager) UpdateMessageMetadata(ctx context.Context, messageID in
|
||||
meta["task_status"] = taskStatus
|
||||
}
|
||||
|
||||
if m.pool != nil {
|
||||
if m.pool != nil || hasUoW(ctx) {
|
||||
metaJSON, _ := json.Marshal(meta)
|
||||
_, err := m.pool.Exec(ctx, `
|
||||
_, err := execChat(ctx, m.pool, `
|
||||
UPDATE chat_messages
|
||||
SET metadata = COALESCE(metadata, '{}'::jsonb) || $1::jsonb
|
||||
WHERE id = $2
|
||||
@@ -63,12 +68,12 @@ func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messa
|
||||
meta["task_status"] = taskStatus
|
||||
}
|
||||
|
||||
if m.pool != nil {
|
||||
if m.pool != nil || hasUoW(ctx) {
|
||||
metaJSON, err := json.Marshal(meta)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal metadata failed: %w", err)
|
||||
}
|
||||
tag, err := m.pool.Exec(ctx, `
|
||||
tag, err := execChat(ctx, m.pool, `
|
||||
UPDATE chat_messages
|
||||
SET content = $1,
|
||||
metadata = COALESCE(metadata, '{}'::jsonb) || $2::jsonb
|
||||
@@ -102,3 +107,22 @@ func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messa
|
||||
}
|
||||
return errors.New("message not found")
|
||||
}
|
||||
|
||||
func hasUoW(ctx context.Context) bool {
|
||||
_, ok := uow.ScopeFrom(ctx)
|
||||
return ok
|
||||
}
|
||||
|
||||
func execChat(ctx context.Context, pool *pgxpool.Pool, q string, args ...any) (pgconn.CommandTag, error) {
|
||||
if scope, ok := uow.ScopeFrom(ctx); ok {
|
||||
tx, err := uow.AsPG(scope)
|
||||
if err != nil {
|
||||
return pgconn.CommandTag{}, err
|
||||
}
|
||||
return tx.Exec(ctx, q, args...)
|
||||
}
|
||||
if pool == nil {
|
||||
return pgconn.CommandTag{}, errors.New("ctxmgr: database required")
|
||||
}
|
||||
return pool.Exec(ctx, q, args...)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user