Fix interrupt handling for suspended tasks

Previously, task suspension did not properly cancel running task execution,
causing tasks to continue processing even after being marked as suspended.

Key changes:
• Add TaskSuspendedEvent to distinguish suspension from normal task events
• Subscribe to suspension events in TaskReconciler to cancel active tasks
• Resume tasks when new messages arrive by resetting DesiredPhase to running
• Check DesiredPhase instead of Phase when computing task status

This ensures tasks can be properly interrupted and later resumed without
leaving zombie executions running in the background.

Co-authored-by: construct-agent <noreply@construct.sh>

Author: Furisto

backend/agent/conv.go

@@ -581,6 +581,8 @@ func ConvertMemoryMessageToProto(m *memory.Message) (*v1.Message, error) {
 			InputTokens:      m.Usage.InputTokens,
 			OutputTokens:     m.Usage.OutputTokens,
 			CacheWriteTokens: m.Usage.CacheWriteTokens,
+			CacheReadTokens:  m.Usage.CacheReadTokens,
+			Cost:             m.Usage.Cost,
 		}
 	}
 

backend/agent/task_reconciler.go

@@ -103,12 +103,20 @@ func (r *TaskReconciler) Run(ctx context.Context) error {
 		}()
 	}
 
-	sub := event.Subscribe(r.bus, func(ctx context.Context, e event.TaskEvent) {
+	taskEventSub := event.Subscribe(r.bus, func(ctx context.Context, e event.TaskEvent) {
 		r.queue.Add(e.TaskID)
 	}, nil)
 
+	taskSuspendedEventSub := event.Subscribe(r.bus, func(ctx context.Context, e event.TaskSuspendedEvent) {
+		cancel, ok := r.runningTasks.Get(e.TaskID)
+		if ok {
+			cancel()
+		}
+	}, nil)
+
 	<-ctx.Done()
-	sub.Unsubscribe()
+	taskEventSub.Unsubscribe()
+	taskSuspendedEventSub.Unsubscribe()
 
 	r.queue.ShutDownWithDrain()
 
@@ -202,7 +210,6 @@ func (r *TaskReconciler) reconcile(ctx context.Context, taskID uuid.UUID) (Resul
 		return Result{}, fmt.Errorf("failed to fetch messages: %w", err)
 	}
 
-	// Trigger title generation if needed
 	if shouldGenerateTitle(task, messages) {
 		go r.generateTitleAsync(taskID)
 	}
@@ -256,7 +263,7 @@ func (r *TaskReconciler) fetchTaskWithAgent(ctx context.Context, taskID uuid.UUI
 
 // computeStatus analyzes the message history and determines what action to take
 func (r *TaskReconciler) computeStatus(task *memory.Task, messages []*memory.Message) (*TaskStatus, error) {
-	if task.Phase == types.TaskPhaseSuspended {
+	if task.DesiredPhase == types.TaskPhaseSuspended {
 		return &TaskStatus{Phase: TaskPhaseSuspended}, nil
 	}
 

backend/api/message.go

@@ -47,6 +47,13 @@ func (h *MessageHandler) CreateMessage(ctx context.Context, req *connect.Request
 			return nil, err
 		}
 
+		if task.DesiredPhase == types.TaskPhaseSuspended {
+			_, err = tx.Task.UpdateOneID(taskID).SetDesiredPhase(types.TaskPhaseRunning).Save(ctx)
+			if err != nil {
+				return nil, err
+			}
+		}
+
 		return tx.Message.Create().
 			SetTask(task).
 			SetContent(conv.ConvertProtoContentToMemory(req.Msg.Content)).

backend/api/task.go

@@ -310,7 +310,7 @@ func (h *TaskHandler) SuspendTask(ctx context.Context, req *connect.Request[v1.S
 		return nil, apiError(err)
 	}
 
-	event.Publish(h.eventBus, event.TaskEvent{
+	event.Publish(h.eventBus, event.TaskSuspendedEvent{
 		TaskID: taskID,
 	})
 	return connect.NewResponse(&v1.SuspendTaskResponse{}), nil

backend/event/events.go

@@ -8,6 +8,12 @@ type TaskEvent struct {
 
 func (TaskEvent) Event() {}
 
+type TaskSuspendedEvent struct {
+	TaskID uuid.UUID
+}
+
+func (TaskSuspendedEvent) Event() {}
+
 type MessageEvent struct {
 	MessageID uuid.UUID
 	TaskID    uuid.UUID

shared/services.go

@@ -199,7 +199,7 @@ func (u *DefaultUserInfo) ConstructLogDir() (string, error) {
 	default:
 		logDir = filepath.Join(xdg.StateHome, "construct")
 	}
-	
+
 	if err := u.fs.MkdirAll(logDir, 0700); err != nil {
 		return "", fmt.Errorf("failed to create log directory: %w", err)
 	}