feat: implement file upload handler and related functionality
- Added file upload handler to process both multipart and raw file uploads. - Implemented parsing logic for upload requests, including handling file metadata. - Introduced SaveFileDecodedInput structure for handling decoded file uploads. - Created unit tests for file upload parsing and validation. feat: add metadata retry configuration and functionality - Introduced MetadataRetryConfig to the application configuration. - Implemented MetadataRetryer to handle retrying metadata extraction for thoughts. - Added new tool for retrying failed metadata extractions. - Updated thought metadata structure to include status and timestamps for metadata processing. fix: enhance metadata normalization and error handling - Updated metadata normalization functions to track status and errors. - Improved handling of metadata extraction failures during thought updates and captures. - Ensured that metadata status is correctly set during various operations. refactor: streamline file saving logic in FilesTool - Refactored Save method in FilesTool to utilize new SaveDecoded method. - Simplified project and thought ID resolution logic during file saving.
This commit is contained in:
206
internal/tools/metadata_retry.go
Normal file
206
internal/tools/metadata_retry.go
Normal file
@@ -0,0 +1,206 @@
|
||||
package tools
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/modelcontextprotocol/go-sdk/mcp"
|
||||
"golang.org/x/sync/semaphore"
|
||||
|
||||
"git.warky.dev/wdevs/amcs/internal/ai"
|
||||
"git.warky.dev/wdevs/amcs/internal/config"
|
||||
"git.warky.dev/wdevs/amcs/internal/metadata"
|
||||
"git.warky.dev/wdevs/amcs/internal/session"
|
||||
"git.warky.dev/wdevs/amcs/internal/store"
|
||||
thoughttypes "git.warky.dev/wdevs/amcs/internal/types"
|
||||
)
|
||||
|
||||
const metadataRetryConcurrency = 4
|
||||
|
||||
type MetadataRetryer struct {
|
||||
backgroundCtx context.Context
|
||||
store *store.DB
|
||||
provider ai.Provider
|
||||
capture config.CaptureConfig
|
||||
sessions *session.ActiveProjects
|
||||
metadataTimeout time.Duration
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
type RetryMetadataTool struct {
|
||||
retryer *MetadataRetryer
|
||||
}
|
||||
|
||||
type RetryMetadataInput struct {
|
||||
Project string `json:"project,omitempty" jsonschema:"optional project name or id to scope the retry"`
|
||||
Limit int `json:"limit,omitempty" jsonschema:"maximum number of thoughts to process in one call; defaults to 100"`
|
||||
IncludeArchived bool `json:"include_archived,omitempty" jsonschema:"whether to include archived thoughts; defaults to false"`
|
||||
OlderThanDays int `json:"older_than_days,omitempty" jsonschema:"only retry thoughts whose last metadata attempt was at least N days ago; 0 means no restriction"`
|
||||
DryRun bool `json:"dry_run,omitempty" jsonschema:"report counts without retrying metadata extraction"`
|
||||
}
|
||||
|
||||
type RetryMetadataFailure struct {
|
||||
ID string `json:"id"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
|
||||
type RetryMetadataOutput struct {
|
||||
Scanned int `json:"scanned"`
|
||||
Retried int `json:"retried"`
|
||||
Updated int `json:"updated"`
|
||||
Skipped int `json:"skipped"`
|
||||
Failed int `json:"failed"`
|
||||
DryRun bool `json:"dry_run"`
|
||||
Failures []RetryMetadataFailure `json:"failures,omitempty"`
|
||||
}
|
||||
|
||||
func NewMetadataRetryer(backgroundCtx context.Context, db *store.DB, provider ai.Provider, capture config.CaptureConfig, metadataTimeout time.Duration, sessions *session.ActiveProjects, logger *slog.Logger) *MetadataRetryer {
|
||||
if backgroundCtx == nil {
|
||||
backgroundCtx = context.Background()
|
||||
}
|
||||
return &MetadataRetryer{
|
||||
backgroundCtx: backgroundCtx,
|
||||
store: db,
|
||||
provider: provider,
|
||||
capture: capture,
|
||||
sessions: sessions,
|
||||
metadataTimeout: metadataTimeout,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
func NewRetryMetadataTool(retryer *MetadataRetryer) *RetryMetadataTool {
|
||||
return &RetryMetadataTool{retryer: retryer}
|
||||
}
|
||||
|
||||
func (t *RetryMetadataTool) Handle(ctx context.Context, req *mcp.CallToolRequest, in RetryMetadataInput) (*mcp.CallToolResult, RetryMetadataOutput, error) {
|
||||
return t.retryer.Handle(ctx, req, in)
|
||||
}
|
||||
|
||||
func (r *MetadataRetryer) QueueThought(id uuid.UUID) {
|
||||
go func() {
|
||||
attemptCtx := r.backgroundCtx
|
||||
if r.metadataTimeout > 0 {
|
||||
var cancel context.CancelFunc
|
||||
attemptCtx, cancel = context.WithTimeout(r.backgroundCtx, r.metadataTimeout)
|
||||
defer cancel()
|
||||
}
|
||||
if _, err := r.retryOne(attemptCtx, id); err != nil {
|
||||
r.logger.Warn("background metadata retry failed", slog.String("thought_id", id.String()), slog.String("error", err.Error()))
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (r *MetadataRetryer) Handle(ctx context.Context, req *mcp.CallToolRequest, in RetryMetadataInput) (*mcp.CallToolResult, RetryMetadataOutput, error) {
|
||||
limit := in.Limit
|
||||
if limit <= 0 {
|
||||
limit = 100
|
||||
}
|
||||
|
||||
project, err := resolveProject(ctx, r.store, r.sessions, req, in.Project, false)
|
||||
if err != nil {
|
||||
return nil, RetryMetadataOutput{}, err
|
||||
}
|
||||
|
||||
var projectID *uuid.UUID
|
||||
if project != nil {
|
||||
projectID = &project.ID
|
||||
}
|
||||
|
||||
thoughts, err := r.store.ListThoughtsPendingMetadataRetry(ctx, limit, projectID, in.IncludeArchived, in.OlderThanDays)
|
||||
if err != nil {
|
||||
return nil, RetryMetadataOutput{}, err
|
||||
}
|
||||
|
||||
out := RetryMetadataOutput{
|
||||
Scanned: len(thoughts),
|
||||
DryRun: in.DryRun,
|
||||
}
|
||||
if in.DryRun || len(thoughts) == 0 {
|
||||
return nil, out, nil
|
||||
}
|
||||
|
||||
sem := semaphore.NewWeighted(metadataRetryConcurrency)
|
||||
var mu sync.Mutex
|
||||
var wg sync.WaitGroup
|
||||
|
||||
for _, thought := range thoughts {
|
||||
if ctx.Err() != nil {
|
||||
break
|
||||
}
|
||||
if err := sem.Acquire(ctx, 1); err != nil {
|
||||
break
|
||||
}
|
||||
|
||||
wg.Add(1)
|
||||
go func(thought thoughttypes.Thought) {
|
||||
defer wg.Done()
|
||||
defer sem.Release(1)
|
||||
|
||||
mu.Lock()
|
||||
out.Retried++
|
||||
mu.Unlock()
|
||||
|
||||
updated, err := r.retryOne(ctx, thought.ID)
|
||||
if err != nil {
|
||||
mu.Lock()
|
||||
out.Failures = append(out.Failures, RetryMetadataFailure{ID: thought.ID.String(), Error: err.Error()})
|
||||
mu.Unlock()
|
||||
return
|
||||
}
|
||||
if updated {
|
||||
mu.Lock()
|
||||
out.Updated++
|
||||
mu.Unlock()
|
||||
return
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
out.Skipped++
|
||||
mu.Unlock()
|
||||
}(thought)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
out.Failed = len(out.Failures)
|
||||
|
||||
return nil, out, nil
|
||||
}
|
||||
|
||||
func (r *MetadataRetryer) retryOne(ctx context.Context, id uuid.UUID) (bool, error) {
|
||||
thought, err := r.store.GetThought(ctx, id)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if thought.Metadata.MetadataStatus == metadata.MetadataStatusComplete {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
attemptCtx := ctx
|
||||
if r.metadataTimeout > 0 {
|
||||
var cancel context.CancelFunc
|
||||
attemptCtx, cancel = context.WithTimeout(ctx, r.metadataTimeout)
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
attemptedAt := time.Now().UTC()
|
||||
extracted, extractErr := r.provider.ExtractMetadata(attemptCtx, thought.Content)
|
||||
if extractErr != nil {
|
||||
failedMetadata := metadata.MarkMetadataFailed(thought.Metadata, r.capture, attemptedAt, extractErr)
|
||||
if _, updateErr := r.store.UpdateThoughtMetadata(ctx, thought.ID, failedMetadata); updateErr != nil {
|
||||
return false, updateErr
|
||||
}
|
||||
return false, extractErr
|
||||
}
|
||||
|
||||
completedMetadata := metadata.MarkMetadataComplete(extracted, r.capture, attemptedAt)
|
||||
completedMetadata.Attachments = thought.Metadata.Attachments
|
||||
if _, updateErr := r.store.UpdateThoughtMetadata(ctx, thought.ID, completedMetadata); updateErr != nil {
|
||||
return false, updateErr
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
Reference in New Issue
Block a user