dreampiler/kandev
1
0
Fork 0
Code Issues 40 Pull Requests Actions 43 Packages Projects Releases Wiki Activity

feat: rename agent_sessions to task_sessions

Browse Source
This commit is contained in:
Carlos Florêncio
2026-01-15 20:31:39 +00:00
parent 9fda624874
commit 712610e428
41 changed files with 486 additions and 486 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/cmd/kandev/main.go
+16 -16
View File
@@ -423,10 +423,10 @@ func main() {
// Set up historical logs provider for task subscriptions
// Uses messages instead of execution logs - all agent messages are now messages
gateway.Hub.SetHistoricalLogsProvider(func(ctx context.Context, taskID string) ([]*ws.Message, error) {
var session *models.AgentSession
session, err = taskRepo.GetActiveAgentSessionByTaskID(ctx, taskID)
var session *models.TaskSession
session, err = taskRepo.GetActiveTaskSessionByTaskID(ctx, taskID)
if err != nil {
session, err = taskRepo.GetAgentSessionByTaskID(ctx, taskID)
session, err = taskRepo.GetTaskSessionByTaskID(ctx, taskID)
if err != nil {
return nil, nil
}
@@ -442,7 +442,7 @@ func main() {
action := ws.ActionMessageAdded
payload := map[string]interface{}{
"message_id": message.ID,
"agent_session_id": message.AgentSessionID,
"task_session_id": message.TaskSessionID,
"task_id": message.TaskID,
"author_type": string(message.AuthorType),
"author_id": message.AuthorID,
@@ -462,7 +462,7 @@ func main() {
}
// Also send current git status if available
session, err = taskRepo.GetActiveAgentSessionByTaskID(ctx, taskID)
session, err = taskRepo.GetActiveTaskSessionByTaskID(ctx, taskID)
if err == nil && session != nil && session.Metadata != nil {
if gitStatus, ok := session.Metadata["git_status"]; ok {
// Add task_id to the git status data
@@ -521,7 +521,7 @@ func main() {
zap.String("task_id", taskID),
zap.String("agent_id", agentID))
session, err := taskRepo.GetActiveAgentSessionByTaskID(ctx, taskID)
session, err := taskRepo.GetActiveTaskSessionByTaskID(ctx, taskID)
if err != nil {
log.Error("failed to resolve active agent session for input request",
zap.String("task_id", taskID),
@@ -531,7 +531,7 @@ func main() {
// Create a message from the agent
messageRecord, err := taskSvc.CreateMessage(ctx, &taskservice.CreateMessageRequest{
AgentSessionID: session.ID,
TaskSessionID: session.ID,
TaskID: taskID,
Content: message,
AuthorType: "agent",
@@ -548,7 +548,7 @@ func main() {
// Broadcast message.added notification to task subscribers
notification, _ := ws.NewNotification(ws.ActionMessageAdded, map[string]interface{}{
"task_id": taskID,
"agent_session_id": session.ID,
"task_session_id": session.ID,
"message": messageRecord.ToAPI(),
"requests_input": true,
})
@@ -581,7 +581,7 @@ func main() {
payload := map[string]interface{}{
"task_id": taskID,
"agent_session_id": agentSessionID,
"task_session_id": agentSessionID,
"message_id": data["message_id"],
"author_type": data["author_type"],
"author_id": data["author_id"],
@@ -618,7 +618,7 @@ func main() {
}
payload := map[string]interface{}{
"message_id": data["message_id"],
"agent_session_id": agentSessionID,
"task_session_id": agentSessionID,
"task_id": taskID,
"author_type": data["author_type"],
"author_id": data["author_id"],
@@ -645,13 +645,13 @@ func main() {
}
// Subscribe to agent_session.state_changed events and broadcast to task subscribers
_, err = eventBus.Subscribe(events.AgentSessionStateChanged, func(ctx context.Context, event *bus.Event) error {
_, err = eventBus.Subscribe(events.TaskSessionStateChanged, func(ctx context.Context, event *bus.Event) error {
data := event.Data
taskID, _ := data["task_id"].(string)
if taskID == "" {
return nil
}
notification, err := ws.NewNotification(ws.ActionAgentSessionStateChanged, data)
notification, err := ws.NewNotification(ws.ActionTaskSessionStateChanged, data)
if err != nil {
log.Error("Failed to create agent_session.state_changed notification", zap.Error(err))
return nil
@@ -1023,7 +1023,7 @@ type messageCreatorAdapter struct {
// CreateAgentMessage creates a message with author_type="agent"
func (a *messageCreatorAdapter) CreateAgentMessage(ctx context.Context, taskID, content, agentSessionID string) error {
_, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{
AgentSessionID: agentSessionID,
TaskSessionID: agentSessionID,
TaskID: taskID,
Content: content,
AuthorType: "agent",
@@ -1034,7 +1034,7 @@ func (a *messageCreatorAdapter) CreateAgentMessage(ctx context.Context, taskID,
// CreateUserMessage creates a message with author_type="user"
func (a *messageCreatorAdapter) CreateUserMessage(ctx context.Context, taskID, content, agentSessionID string) error {
_, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{
AgentSessionID: agentSessionID,
TaskSessionID: agentSessionID,
TaskID: taskID,
Content: content,
AuthorType: "user",
@@ -1061,7 +1061,7 @@ func (a *messageCreatorAdapter) CreateToolCallMessage(ctx context.Context, taskI
}
_, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{
AgentSessionID: agentSessionID,
TaskSessionID: agentSessionID,
TaskID: taskID,
Content: title,
AuthorType: "agent",
@@ -1079,7 +1079,7 @@ func (a *messageCreatorAdapter) UpdateToolCallMessage(ctx context.Context, taskI
// CreateSessionMessage creates a message for non-chat session updates (status/progress/error/etc).
func (a *messageCreatorAdapter) CreateSessionMessage(ctx context.Context, taskID, content, agentSessionID, messageType string, metadata map[string]interface{}, requestsInput bool) error {
_, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{
AgentSessionID: agentSessionID,
TaskSessionID: agentSessionID,
TaskID: taskID,
Content: content,
AuthorType: "agent",
apps/backend/dev/seed.sql
+2 -2
View File
@@ -77,7 +77,7 @@ VALUES
('comment-004', 'task-006', 'user', 'dev-user', 'Need to add support for tool call rendering in the chat panel', 'message', '{}', 0, '', datetime('now', '-1 day')),
('comment-005', 'task-006', 'agent', 'augment-agent', 'I''ve implemented the tool call comment type with proper rendering. The changes include:\n\n1. Added `type` field to comments\n2. Added `metadata` for tool call details\n3. Updated the frontend to render tool calls differently', 'message', '{}', 0, '', datetime('now', '-20 hours'));
-- Sample agent session
INSERT OR REPLACE INTO agent_sessions (id, task_id, agent_instance_id, agent_type, state, progress, error_message, metadata, started_at, completed_at, updated_at)
-- Sample task session
INSERT OR REPLACE INTO task_sessions (id, task_id, agent_instance_id, agent_type, state, progress, error_message, metadata, started_at, completed_at, updated_at)
VALUES
('session-001', 'task-005', 'agent-instance-123', 'augment-agent', 'RUNNING', 0.0, '', '{}', datetime('now', '-1 hour'), NULL, datetime('now'));
apps/backend/internal/agent/settings/controller/controller.go
+2 -2
View File
@@ -29,7 +29,7 @@ type Controller struct {
}
type SessionChecker interface {
HasActiveAgentSessionsByAgentProfile(ctx context.Context, agentProfileID string) (bool, error)
HasActiveTaskSessionsByAgentProfile(ctx context.Context, agentProfileID string) (bool, error)
}
func NewController(repo store.Repository, discoveryRegistry *discovery.Registry, sessionChecker SessionChecker, log *logger.Logger) *Controller {
@@ -274,7 +274,7 @@ func (c *Controller) DeleteProfile(ctx context.Context, id string) (*dto.AgentPr
return nil, err
}
if c.sessionChecker != nil {
active, err := c.sessionChecker.HasActiveAgentSessionsByAgentProfile(ctx, id)
active, err := c.sessionChecker.HasActiveTaskSessionsByAgentProfile(ctx, id)
if err != nil {
return nil, err
}
apps/backend/internal/events/types.go
+2 -2
View File
@@ -36,9 +36,9 @@ const (
MessageUpdated = "message.updated"
)
// Event types for agent sessions
// Event types for task sessions
const (
AgentSessionStateChanged = "agent_session.state_changed"
TaskSessionStateChanged = "task_session.state_changed"
)
// Event types for repositories
apps/backend/internal/orchestrator/controller/controller.go
+2 -2
View File
@@ -71,7 +71,7 @@ func (c *Controller) StartTask(ctx context.Context, req dto.StartTaskRequest) (d
Success: true,
TaskID: execution.TaskID,
AgentInstanceID: execution.AgentInstanceID,
AgentSessionID: execution.SessionID,
TaskSessionID: execution.SessionID,
State: string(execution.SessionState),
}
@@ -153,7 +153,7 @@ func (c *Controller) GetTaskExecution(ctx context.Context, req dto.GetTaskExecut
TaskID: execution.TaskID,
AgentInstanceID: execution.AgentInstanceID,
AgentProfileID: execution.AgentProfileID,
AgentSessionID: execution.SessionID,
TaskSessionID: execution.SessionID,
State: string(execution.SessionState),
Progress: execution.Progress,
StartedAt: execution.StartedAt.Format("2006-01-02T15:04:05Z"),
apps/backend/internal/orchestrator/dto/dto.go
+2 -2
View File
@@ -52,7 +52,7 @@ type StartTaskResponse struct {
Success bool `json:"success"`
TaskID string `json:"task_id"`
AgentInstanceID string `json:"agent_instance_id"`
AgentSessionID string `json:"agent_session_id,omitempty"`
TaskSessionID string `json:"agent_session_id,omitempty"`
State string `json:"state"`
WorktreePath *string `json:"worktree_path,omitempty"`
WorktreeBranch *string `json:"worktree_branch,omitempty"`
@@ -119,7 +119,7 @@ type TaskExecutionResponse struct {
TaskID string `json:"task_id"`
AgentInstanceID string `json:"agent_instance_id,omitempty"`
AgentProfileID string `json:"agent_profile_id,omitempty"`
AgentSessionID string `json:"agent_session_id,omitempty"`
TaskSessionID string `json:"agent_session_id,omitempty"`
State string `json:"state,omitempty"`
Progress int `json:"progress,omitempty"`
StartedAt string `json:"started_at,omitempty"`
apps/backend/internal/orchestrator/executor/executor.go
+34 -34
View File
@@ -92,7 +92,7 @@ type TaskExecution struct {
AgentInstanceID string
AgentProfileID string
StartedAt time.Time
SessionState v1.AgentSessionState
SessionState v1.TaskSessionState
Progress int
LastUpdate time.Time
// SessionID is the database ID of the agent session
@@ -102,8 +102,8 @@ type TaskExecution struct {
WorktreeBranch string
}
// FromAgentSession converts a models.AgentSession to TaskExecution
func FromAgentSession(s *models.AgentSession) *TaskExecution {
// FromTaskSession converts a models.TaskSession to TaskExecution
func FromTaskSession(s *models.TaskSession) *TaskExecution {
return &TaskExecution{
TaskID: s.TaskID,
AgentInstanceID: s.AgentInstanceID,
@@ -116,9 +116,9 @@ func FromAgentSession(s *models.AgentSession) *TaskExecution {
}
}
// agentSessionStateToV1 converts models.AgentSessionState to v1.AgentSessionState
func agentSessionStateToV1(state models.AgentSessionState) v1.AgentSessionState {
return v1.AgentSessionState(state)
// agentSessionStateToV1 converts models.TaskSessionState to v1.TaskSessionState
func agentSessionStateToV1(state models.TaskSessionState) v1.TaskSessionState {
return v1.TaskSessionState(state)
}
// Executor manages agent execution for tasks
@@ -170,7 +170,7 @@ func (e *Executor) SetWorktreeEnabled(enabled bool) {
// LoadActiveSessionsFromDB loads active agent sessions from the database into memory
// This should be called on startup to restore state after a restart
func (e *Executor) LoadActiveSessionsFromDB(ctx context.Context) error {
sessions, err := e.repo.ListActiveAgentSessions(ctx)
sessions, err := e.repo.ListActiveTaskSessions(ctx)
if err != nil {
return err
}
@@ -179,7 +179,7 @@ func (e *Executor) LoadActiveSessionsFromDB(ctx context.Context) error {
defer e.mu.Unlock()
for _, session := range sessions {
e.executions[session.TaskID] = FromAgentSession(session)
e.executions[session.TaskID] = FromTaskSession(session)
e.logger.Info("restored agent session from database",
zap.String("task_id", session.TaskID),
zap.String("session_id", session.ID),
@@ -218,7 +218,7 @@ func (e *Executor) SyncWithRecoveredInstances(ctx context.Context, instances []R
AgentInstanceID: inst.InstanceID,
AgentProfileID: inst.AgentProfileID,
StartedAt: time.Now(), // We don't know exact start time
SessionState: v1.AgentSessionStateRunning,
SessionState: v1.TaskSessionStateRunning,
Progress: 0,
LastUpdate: time.Now(),
}
@@ -314,7 +314,7 @@ func (e *Executor) ExecuteWithProfile(ctx context.Context, task *v1.Task, agentP
// Create agent session in database
sessionID := uuid.New().String()
now := time.Now().UTC()
session := &models.AgentSession{
session := &models.TaskSession{
ID: sessionID,
TaskID: task.ID,
AgentInstanceID: resp.AgentInstanceID,
@@ -324,13 +324,13 @@ func (e *Executor) ExecuteWithProfile(ctx context.Context, task *v1.Task, agentP
BaseBranch: baseBranch,
WorktreePath: resp.WorktreePath,
WorktreeBranch: resp.WorktreeBranch,
State: models.AgentSessionStateStarting,
State: models.TaskSessionStateStarting,
Progress: 0,
StartedAt: now,
UpdatedAt: now,
}
if err := e.repo.CreateAgentSession(ctx, session); err != nil {
if err := e.repo.CreateTaskSession(ctx, session); err != nil {
e.logger.Error("failed to persist agent session",
zap.String("task_id", task.ID),
zap.Error(err))
@@ -343,7 +343,7 @@ func (e *Executor) ExecuteWithProfile(ctx context.Context, task *v1.Task, agentP
AgentInstanceID: resp.AgentInstanceID,
AgentProfileID: agentProfileID,
StartedAt: now,
SessionState: v1.AgentSessionStateStarting,
SessionState: v1.TaskSessionStateStarting,
Progress: 0,
LastUpdate: now,
SessionID: sessionID,
@@ -391,7 +391,7 @@ func (e *Executor) Stop(ctx context.Context, taskID string, reason string, force
// Update execution status in memory and database regardless of agent stop result
e.mu.Lock()
if exec, ok := e.executions[taskID]; ok {
exec.SessionState = v1.AgentSessionStateCancelled
exec.SessionState = v1.TaskSessionStateCancelled
exec.LastUpdate = time.Now()
}
delete(e.executions, taskID) // Remove from cache so GetExecution returns false
@@ -399,7 +399,7 @@ func (e *Executor) Stop(ctx context.Context, taskID string, reason string, force
// Update database
if execution.SessionID != "" {
if dbErr := e.repo.UpdateAgentSessionState(ctx, execution.SessionID, models.AgentSessionStateCancelled, reason); dbErr != nil {
if dbErr := e.repo.UpdateTaskSessionState(ctx, execution.SessionID, models.TaskSessionStateCancelled, reason); dbErr != nil {
e.logger.Error("failed to update agent session status in database",
zap.String("task_id", taskID),
zap.String("session_id", execution.SessionID),
@@ -438,7 +438,7 @@ func (e *Executor) getOrLoadExecution(ctx context.Context, taskID string) (*Task
}
// Try to load from database
session, err := e.repo.GetActiveAgentSessionByTaskID(ctx, taskID)
session, err := e.repo.GetActiveTaskSessionByTaskID(ctx, taskID)
if err != nil {
// Check if it's a "not found" error
if strings.Contains(err.Error(), "no active agent session") {
@@ -448,7 +448,7 @@ func (e *Executor) getOrLoadExecution(ctx context.Context, taskID string) (*Task
}
// Cache it in memory
execution = FromAgentSession(session)
execution = FromTaskSession(session)
e.mu.Lock()
e.executions[taskID] = execution
e.mu.Unlock()
@@ -489,7 +489,7 @@ func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) {
// Also update DB if we have a session
if execution.SessionID != "" {
_ = e.repo.UpdateAgentSessionState(ctx, execution.SessionID, models.AgentSessionStateCancelled, "agent process stopped")
_ = e.repo.UpdateTaskSessionState(ctx, execution.SessionID, models.TaskSessionStateCancelled, "agent process stopped")
}
return nil, false
}
@@ -500,7 +500,7 @@ func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) {
}
// Try to load from database
session, err := e.repo.GetActiveAgentSessionByTaskID(ctx, taskID)
session, err := e.repo.GetActiveTaskSessionByTaskID(ctx, taskID)
if err != nil {
return nil, false
}
@@ -512,12 +512,12 @@ func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) {
e.logger.Info("stale agent session detected - agent not running",
zap.String("task_id", taskID),
zap.String("session_id", session.ID))
_ = e.repo.UpdateAgentSessionState(ctx, session.ID, models.AgentSessionStateCancelled, "agent not running after backend restart")
_ = e.repo.UpdateTaskSessionState(ctx, session.ID, models.TaskSessionStateCancelled, "agent not running after backend restart")
return nil, false
}
// Cache it in memory
execution = FromAgentSession(session)
execution = FromTaskSession(session)
e.mu.Lock()
e.executions[taskID] = execution
e.mu.Unlock()
@@ -542,7 +542,7 @@ func (e *Executor) GetExecutionWithContext(ctx context.Context, taskID string) (
// Also update DB if we have a session
if execution.SessionID != "" {
_ = e.repo.UpdateAgentSessionState(ctx, execution.SessionID, models.AgentSessionStateCancelled, "agent process stopped")
_ = e.repo.UpdateTaskSessionState(ctx, execution.SessionID, models.TaskSessionStateCancelled, "agent process stopped")
}
return nil, false
}
@@ -552,7 +552,7 @@ func (e *Executor) GetExecutionWithContext(ctx context.Context, taskID string) (
}
// Try to load from database
session, err := e.repo.GetActiveAgentSessionByTaskID(ctx, taskID)
session, err := e.repo.GetActiveTaskSessionByTaskID(ctx, taskID)
if err != nil {
return nil, false
}
@@ -562,12 +562,12 @@ func (e *Executor) GetExecutionWithContext(ctx context.Context, taskID string) (
e.logger.Info("stale agent session detected - agent not running",
zap.String("task_id", taskID),
zap.String("session_id", session.ID))
_ = e.repo.UpdateAgentSessionState(ctx, session.ID, models.AgentSessionStateCancelled, "agent not running after backend restart")
_ = e.repo.UpdateTaskSessionState(ctx, session.ID, models.TaskSessionStateCancelled, "agent not running after backend restart")
return nil, false
}
// Cache it in memory
execution = FromAgentSession(session)
execution = FromTaskSession(session)
e.mu.Lock()
e.executions[taskID] = execution
e.mu.Unlock()
@@ -602,7 +602,7 @@ func (e *Executor) CanExecute() bool {
}
// UpdateProgress updates the progress of an execution (called when ACP progress messages are received)
func (e *Executor) UpdateProgress(ctx context.Context, taskID string, progress int, state v1.AgentSessionState) {
func (e *Executor) UpdateProgress(ctx context.Context, taskID string, progress int, state v1.TaskSessionState) {
e.mu.Lock()
var sessionID string
if exec, ok := e.executions[taskID]; ok {
@@ -621,13 +621,13 @@ func (e *Executor) UpdateProgress(ctx context.Context, taskID string, progress i
// Update database asynchronously (don't block on this)
if sessionID != "" {
go func() {
session, err := e.repo.GetAgentSession(ctx, sessionID)
session, err := e.repo.GetTaskSession(ctx, sessionID)
if err != nil {
return
}
session.Progress = progress
session.State = models.AgentSessionState(state)
if err := e.repo.UpdateAgentSession(ctx, session); err != nil {
session.State = models.TaskSessionState(state)
if err := e.repo.UpdateTaskSession(ctx, session); err != nil {
e.logger.Error("failed to update agent session progress in database",
zap.String("session_id", sessionID),
zap.Error(err))
@@ -637,7 +637,7 @@ func (e *Executor) UpdateProgress(ctx context.Context, taskID string, progress i
}
// MarkCompleted marks an execution as completed
func (e *Executor) MarkCompleted(ctx context.Context, taskID string, state v1.AgentSessionState) {
func (e *Executor) MarkCompleted(ctx context.Context, taskID string, state v1.TaskSessionState) {
e.mu.Lock()
var sessionID string
if exec, ok := e.executions[taskID]; ok {
@@ -654,8 +654,8 @@ func (e *Executor) MarkCompleted(ctx context.Context, taskID string, state v1.Ag
// Update database
if sessionID != "" {
dbState := models.AgentSessionState(state)
if err := e.repo.UpdateAgentSessionState(ctx, sessionID, dbState, ""); err != nil {
dbState := models.TaskSessionState(state)
if err := e.repo.UpdateTaskSessionState(ctx, sessionID, dbState, ""); err != nil {
e.logger.Error("failed to update agent session status in database",
zap.String("session_id", sessionID),
zap.Error(err))
@@ -664,7 +664,7 @@ func (e *Executor) MarkCompleted(ctx context.Context, taskID string, state v1.Ag
}
// UpdateExecutionState updates the in-memory execution state for a task.
func (e *Executor) UpdateExecutionState(taskID string, state v1.AgentSessionState) {
func (e *Executor) UpdateExecutionState(taskID string, state v1.TaskSessionState) {
e.mu.Lock()
defer e.mu.Unlock()
@@ -698,7 +698,7 @@ func (e *Executor) GetActiveSessionID(ctx context.Context, taskID string) (strin
}
// Try to load from database
session, err := e.repo.GetActiveAgentSessionByTaskID(ctx, taskID)
session, err := e.repo.GetActiveTaskSessionByTaskID(ctx, taskID)
if err != nil {
return "", err
}
apps/backend/internal/orchestrator/service.go
+16 -16
View File
@@ -323,7 +323,7 @@ func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID s
}
if execution.SessionID != "" {
s.updateAgentSessionState(ctx, taskID, execution.SessionID, models.AgentSessionStateRunning, "", true)
s.updateTaskSessionState(ctx, taskID, execution.SessionID, models.TaskSessionStateRunning, "", true)
if s.messageCreator != nil && task.Description != "" {
if err := s.messageCreator.CreateUserMessage(ctx, taskID, task.Description, execution.SessionID); err != nil {
s.logger.Error("failed to create initial user message",
@@ -363,7 +363,7 @@ func (s *Service) PromptTask(ctx context.Context, taskID string, prompt string)
zap.String("task_id", taskID),
zap.Int("prompt_length", len(prompt)))
if sessionID, err := s.executor.GetActiveSessionID(ctx, taskID); err == nil && sessionID != "" {
s.updateAgentSessionState(ctx, taskID, sessionID, models.AgentSessionStateRunning, "", true)
s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateRunning, "", true)
}
result, err := s.executor.Prompt(ctx, taskID, prompt)
if err != nil {
@@ -563,7 +563,7 @@ func (s *Service) handleACPMessage(ctx context.Context, taskID string, msg *prot
}
if normalized.bumpToRunning {
s.updateAgentSessionState(ctx, taskID, sessionID, models.AgentSessionStateRunning, "", false)
s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateRunning, "", false)
}
}
}
@@ -648,23 +648,23 @@ func buildACPMetadata(msg *protocol.Message) map[string]interface{} {
return metadata
}
func (s *Service) updateAgentSessionState(ctx context.Context, taskID, sessionID string, nextState models.AgentSessionState, errorMessage string, allowWakeFromWaiting bool) {
session, err := s.repo.GetAgentSession(ctx, sessionID)
func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID string, nextState models.TaskSessionState, errorMessage string, allowWakeFromWaiting bool) {
session, err := s.repo.GetTaskSession(ctx, sessionID)
if err != nil {
return
}
if session.State == models.AgentSessionStateWaitingForInput && nextState == models.AgentSessionStateRunning && !allowWakeFromWaiting {
if session.State == models.TaskSessionStateWaitingForInput && nextState == models.TaskSessionStateRunning && !allowWakeFromWaiting {
return
}
oldState := session.State
switch session.State {
case models.AgentSessionStateCompleted, models.AgentSessionStateFailed, models.AgentSessionStateCancelled:
case models.TaskSessionStateCompleted, models.TaskSessionStateFailed, models.TaskSessionStateCancelled:
return
}
if session.State == nextState {
return
}
if err := s.repo.UpdateAgentSessionState(ctx, sessionID, nextState, errorMessage); err != nil {
if err := s.repo.UpdateTaskSessionState(ctx, sessionID, nextState, errorMessage); err != nil {
s.logger.Error("failed to update agent session state",
zap.String("session_id", sessionID),
zap.String("state", string(nextState)),
@@ -676,7 +676,7 @@ func (s *Service) updateAgentSessionState(ctx context.Context, taskID, sessionID
zap.String("old_state", string(oldState)),
zap.String("new_state", string(nextState)))
if s.eventBus != nil {
_ = s.eventBus.Publish(ctx, events.AgentSessionStateChanged, bus.NewEvent(events.AgentSessionStateChanged, "agent-session", map[string]interface{}{
_ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent(events.TaskSessionStateChanged, "agent-session", map[string]interface{}{
"task_id": taskID,
"agent_session_id": sessionID,
"old_state": string(oldState),
@@ -684,7 +684,7 @@ func (s *Service) updateAgentSessionState(ctx context.Context, taskID, sessionID
}))
}
if taskID != "" {
s.executor.UpdateExecutionState(taskID, v1.AgentSessionState(nextState))
s.executor.UpdateExecutionState(taskID, v1.TaskSessionState(nextState))
}
}
@@ -733,7 +733,7 @@ func (s *Service) finalizeAgentReady(taskID, sessionID string) {
}
if sessionID != "" {
s.updateAgentSessionState(ctx, taskID, sessionID, models.AgentSessionStateWaitingForInput, "", false)
s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateWaitingForInput, "", false)
}
task, err := s.taskRepo.GetTask(ctx, taskID)
@@ -783,7 +783,7 @@ func (s *Service) handleInputRequired(ctx context.Context, taskID string, msg *p
// Update session state to WAITING_FOR_INPUT
if sessionID, err := s.executor.GetActiveSessionID(ctx, taskID); err == nil && sessionID != "" {
s.updateAgentSessionState(ctx, taskID, sessionID, models.AgentSessionStateWaitingForInput, "", false)
s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateWaitingForInput, "", false)
}
// Call the input request handler if set
@@ -803,7 +803,7 @@ func (s *Service) handleGitStatusUpdated(ctx context.Context, data watcher.GitSt
zap.String("branch", data.Branch))
// Get the active agent session for this task
session, err := s.repo.GetActiveAgentSessionByTaskID(ctx, data.TaskID)
session, err := s.repo.GetActiveTaskSessionByTaskID(ctx, data.TaskID)
if err != nil {
s.logger.Debug("no active agent session for git status update",
zap.String("task_id", data.TaskID),
@@ -831,7 +831,7 @@ func (s *Service) handleGitStatusUpdated(ctx context.Context, data watcher.GitSt
// Persist to database asynchronously
go func() {
if err := s.repo.UpdateAgentSession(context.Background(), session); err != nil {
if err := s.repo.UpdateTaskSession(context.Background(), session); err != nil {
s.logger.Error("failed to update agent session with git status",
zap.String("task_id", data.TaskID),
zap.String("session_id", session.ID),
@@ -898,7 +898,7 @@ func (s *Service) handleToolCallStarted(ctx context.Context, data watcher.ToolCa
}
if sessionID != "" {
s.updateAgentSessionState(ctx, data.TaskID, sessionID, models.AgentSessionStateRunning, "", false)
s.updateTaskSessionState(ctx, data.TaskID, sessionID, models.TaskSessionStateRunning, "", false)
}
}
}
@@ -916,7 +916,7 @@ func (s *Service) handleToolCallComplete(ctx context.Context, data watcher.ToolC
}
if sessionID != "" {
s.updateAgentSessionState(ctx, data.TaskID, sessionID, models.AgentSessionStateRunning, "", false)
s.updateTaskSessionState(ctx, data.TaskID, sessionID, models.TaskSessionStateRunning, "", false)
}
}
}
apps/backend/internal/task/controller/environment.go
+2 -2
View File
@@ -71,8 +71,8 @@ func (c *EnvironmentController) UpdateEnvironment(ctx context.Context, req dto.U
func (c *EnvironmentController) DeleteEnvironment(ctx context.Context, req dto.DeleteEnvironmentRequest) (dto.SuccessResponse, error) {
if err := c.service.DeleteEnvironment(ctx, req.ID); err != nil {
if errors.Is(err, service.ErrActiveAgentSessions) {
return dto.SuccessResponse{}, ErrActiveAgentSessions
if errors.Is(err, service.ErrActiveTaskSessions) {
return dto.SuccessResponse{}, ErrActiveTaskSessions
}
return dto.SuccessResponse{}, err
}
apps/backend/internal/task/controller/errors.go
+1 -1
View File
@@ -2,4 +2,4 @@ package controller
import "errors"
var ErrActiveAgentSessions = errors.New("active agent sessions exist")
var ErrActiveTaskSessions = errors.New("active task sessions exist")
apps/backend/internal/task/controller/executor.go
+2 -2
View File
@@ -68,8 +68,8 @@ func (c *ExecutorController) UpdateExecutor(ctx context.Context, req dto.UpdateE
func (c *ExecutorController) DeleteExecutor(ctx context.Context, req dto.DeleteExecutorRequest) (dto.SuccessResponse, error) {
if err := c.service.DeleteExecutor(ctx, req.ID); err != nil {
if errors.Is(err, service.ErrActiveAgentSessions) {
return dto.SuccessResponse{}, ErrActiveAgentSessions
if errors.Is(err, service.ErrActiveTaskSessions) {
return dto.SuccessResponse{}, ErrActiveTaskSessions
}
return dto.SuccessResponse{}, err
}
apps/backend/internal/task/controller/message.go
+2 -2
View File
@@ -18,7 +18,7 @@ func NewMessageController(svc *service.Service) *MessageController {
func (c *MessageController) ListMessages(ctx context.Context, req dto.ListMessagesRequest) (dto.ListMessagesResponse, error) {
messages, hasMore, err := c.service.ListMessagesPaginated(ctx, service.ListMessagesRequest{
AgentSessionID: req.AgentSessionID,
TaskSessionID: req.TaskSessionID,
Limit: req.Limit,
Before: req.Before,
After: req.After,
@@ -62,7 +62,7 @@ func (c *MessageController) ListAllMessages(ctx context.Context, sessionID strin
func (c *MessageController) CreateMessage(ctx context.Context, req dto.CreateMessageRequest) (*v1.Message, error) {
message, err := c.service.CreateMessage(ctx, &service.CreateMessageRequest{
AgentSessionID: req.AgentSessionID,
TaskSessionID: req.TaskSessionID,
TaskID: req.TaskID,
Content: req.Content,
AuthorType: req.AuthorType,
apps/backend/internal/task/controller/repository.go
+2 -2
View File
@@ -80,8 +80,8 @@ func (c *RepositoryController) UpdateRepository(ctx context.Context, req dto.Upd
func (c *RepositoryController) DeleteRepository(ctx context.Context, req dto.DeleteRepositoryRequest) (dto.SuccessResponse, error) {
if err := c.service.DeleteRepository(ctx, req.ID); err != nil {
if errors.Is(err, service.ErrActiveAgentSessions) {
return dto.SuccessResponse{}, ErrActiveAgentSessions
if errors.Is(err, service.ErrActiveTaskSessions) {
return dto.SuccessResponse{}, ErrActiveTaskSessions
}
return dto.SuccessResponse{}, err
}
apps/backend/internal/task/dto/requests.go
+2 -2
View File
@@ -266,7 +266,7 @@ type DeleteTaskRequest struct {
}
type ListMessagesRequest struct {
AgentSessionID string
TaskSessionID string
Limit int
Before string
After string
@@ -274,7 +274,7 @@ type ListMessagesRequest struct {
}
type CreateMessageRequest struct {
AgentSessionID string
TaskSessionID string
Content string
AuthorType string
AuthorID string
apps/backend/internal/task/handlers/environment_handlers.go
+2 -2
View File
@@ -142,7 +142,7 @@ func (h *EnvironmentHandlers) httpUpdateEnvironment(c *gin.Context) {
func (h *EnvironmentHandlers) httpDeleteEnvironment(c *gin.Context) {
resp, err := h.controller.DeleteEnvironment(c.Request.Context(), dto.DeleteEnvironmentRequest{ID: c.Param("id")})
if err != nil {
if errors.Is(err, controller.ErrActiveAgentSessions) {
if errors.Is(err, controller.ErrActiveTaskSessions) {
c.JSON(http.StatusConflict, gin.H{"error": "environment is used by an active agent session"})
return
}
@@ -267,7 +267,7 @@ func (h *EnvironmentHandlers) wsDeleteEnvironment(ctx context.Context, msg *ws.M
resp, err := h.controller.DeleteEnvironment(ctx, dto.DeleteEnvironmentRequest{ID: req.ID})
if err != nil {
h.logger.Error("failed to delete environment", zap.Error(err))
if errors.Is(err, controller.ErrActiveAgentSessions) {
if errors.Is(err, controller.ErrActiveTaskSessions) {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "environment is used by an active agent session", nil)
}
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to delete environment", nil)
apps/backend/internal/task/handlers/executor_handlers.go
+2 -2
View File
@@ -145,7 +145,7 @@ func (h *ExecutorHandlers) httpUpdateExecutor(c *gin.Context) {
func (h *ExecutorHandlers) httpDeleteExecutor(c *gin.Context) {
resp, err := h.controller.DeleteExecutor(c.Request.Context(), dto.DeleteExecutorRequest{ID: c.Param("id")})
if err != nil {
if errors.Is(err, controller.ErrActiveAgentSessions) {
if errors.Is(err, controller.ErrActiveTaskSessions) {
c.JSON(http.StatusConflict, gin.H{"error": "executor is used by an active agent session"})
return
}
@@ -273,7 +273,7 @@ func (h *ExecutorHandlers) wsDeleteExecutor(ctx context.Context, msg *ws.Message
resp, err := h.controller.DeleteExecutor(ctx, dto.DeleteExecutorRequest{ID: req.ID})
if err != nil {
h.logger.Error("failed to delete executor", zap.Error(err))
if errors.Is(err, controller.ErrActiveAgentSessions) {
if errors.Is(err, controller.ErrActiveTaskSessions) {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "executor is used by an active agent session", nil)
}
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to delete executor", nil)
apps/backend/internal/task/handlers/message_handlers.go
+7 -7
View File
@@ -94,7 +94,7 @@ func (h *MessageHandlers) httpListMessages(c *gin.Context) {
)
if paginated {
resp, err = h.messageController.ListMessages(c.Request.Context(), dto.ListMessagesRequest{
AgentSessionID: sessionID,
TaskSessionID: sessionID,
Limit: limit,
Before: before,
After: after,
@@ -115,7 +115,7 @@ func (h *MessageHandlers) httpListMessages(c *gin.Context) {
type wsAddMessageRequest struct {
TaskID string `json:"task_id"`
AgentSessionID string `json:"agent_session_id"`
TaskSessionID string `json:"agent_session_id"`
Content string `json:"content"`
AuthorID string `json:"author_id,omitempty"`
}
@@ -125,7 +125,7 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
if req.AgentSessionID == "" {
if req.TaskSessionID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_session_id is required", nil)
}
if req.TaskID == "" {
@@ -156,7 +156,7 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w
}
message, err := h.messageController.CreateMessage(ctx, dto.CreateMessageRequest{
AgentSessionID: req.AgentSessionID,
TaskSessionID: req.TaskSessionID,
TaskID: req.TaskID,
Content: req.Content,
AuthorType: "user",
@@ -181,7 +181,7 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w
}
type wsListMessagesRequest struct {
AgentSessionID string `json:"agent_session_id"`
TaskSessionID string `json:"agent_session_id"`
Limit int `json:"limit"`
Before string `json:"before"`
After string `json:"after"`
@@ -193,7 +193,7 @@ func (h *MessageHandlers) wsListMessages(ctx context.Context, msg *ws.Message) (
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
if req.AgentSessionID == "" {
if req.TaskSessionID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_session_id is required", nil)
}
if req.Before != "" && req.After != "" {
@@ -204,7 +204,7 @@ func (h *MessageHandlers) wsListMessages(ctx context.Context, msg *ws.Message) (
}
resp, err := h.messageController.ListMessages(ctx, dto.ListMessagesRequest{
AgentSessionID: req.AgentSessionID,
TaskSessionID: req.TaskSessionID,
Limit: req.Limit,
Before: req.Before,
After: req.After,
apps/backend/internal/task/handlers/repository_handlers.go
+2 -2
View File
@@ -219,7 +219,7 @@ func (h *RepositoryHandlers) httpUpdateRepository(c *gin.Context) {
func (h *RepositoryHandlers) httpDeleteRepository(c *gin.Context) {
if _, err := h.controller.DeleteRepository(c.Request.Context(), dto.DeleteRepositoryRequest{ID: c.Param("id")}); err != nil {
if errors.Is(err, controller.ErrActiveAgentSessions) {
if errors.Is(err, controller.ErrActiveTaskSessions) {
c.JSON(http.StatusConflict, gin.H{"error": "repository is used by an active agent session"})
return
}
@@ -445,7 +445,7 @@ func (h *RepositoryHandlers) wsDeleteRepository(ctx context.Context, msg *ws.Mes
}
if _, err := h.controller.DeleteRepository(ctx, dto.DeleteRepositoryRequest{ID: req.ID}); err != nil {
h.logger.Error("failed to delete repository", zap.Error(err))
if errors.Is(err, controller.ErrActiveAgentSessions) {
if errors.Is(err, controller.ErrActiveTaskSessions) {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "repository is used by an active agent session", nil)
}
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "Repository not found", nil)
apps/backend/internal/task/models/models.go
+25 -25
View File
@@ -88,10 +88,10 @@ const (
MessageTypeStatus MessageType = "status"
)
// Message represents a message in an agent session
// Message represents a message in a task session
type Message struct {
ID string `json:"id"`
AgentSessionID string `json:"agent_session_id"`
TaskSessionID string `json:"task_session_id"`
TaskID string `json:"task_id,omitempty"`
AuthorType MessageAuthorType `json:"author_type"`
AuthorID string `json:"author_id,omitempty"` // User ID or Agent Instance ID
@@ -110,7 +110,7 @@ func (m *Message) ToAPI() *v1.Message {
}
result := &v1.Message{
ID: m.ID,
AgentSessionID: m.AgentSessionID,
TaskSessionID: m.TaskSessionID,
TaskID: m.TaskID,
AuthorType: string(m.AuthorType),
AuthorID: m.AuthorID,
@@ -123,29 +123,29 @@ func (m *Message) ToAPI() *v1.Message {
return result
}
// AgentSessionState represents the state of an agent session
type AgentSessionState string
// TaskSessionState represents the state of an agent session
type TaskSessionState string
const (
// AgentSessionStateCreated - session created but agent not started
AgentSessionStateCreated AgentSessionState = "CREATED"
// AgentSessionStateStarting - agent is starting up
AgentSessionStateStarting AgentSessionState = "STARTING"
// AgentSessionStateRunning - agent is actively running
AgentSessionStateRunning AgentSessionState = "RUNNING"
// AgentSessionStateWaitingForInput - agent waiting for user input
AgentSessionStateWaitingForInput AgentSessionState = "WAITING_FOR_INPUT"
// AgentSessionStateCompleted - agent finished successfully
AgentSessionStateCompleted AgentSessionState = "COMPLETED"
// AgentSessionStateFailed - agent failed with error
AgentSessionStateFailed AgentSessionState = "FAILED"
// AgentSessionStateCancelled - agent was manually stopped
AgentSessionStateCancelled AgentSessionState = "CANCELLED"
// TaskSessionStateCreated - session created but agent not started
TaskSessionStateCreated TaskSessionState = "CREATED"
// TaskSessionStateStarting - agent is starting up
TaskSessionStateStarting TaskSessionState = "STARTING"
// TaskSessionStateRunning - agent is actively running
TaskSessionStateRunning TaskSessionState = "RUNNING"
// TaskSessionStateWaitingForInput - agent waiting for user input
TaskSessionStateWaitingForInput TaskSessionState = "WAITING_FOR_INPUT"
// TaskSessionStateCompleted - agent finished successfully
TaskSessionStateCompleted TaskSessionState = "COMPLETED"
// TaskSessionStateFailed - agent failed with error
TaskSessionStateFailed TaskSessionState = "FAILED"
// TaskSessionStateCancelled - agent was manually stopped
TaskSessionStateCancelled TaskSessionState = "CANCELLED"
)
// AgentSession represents a persistent agent execution session for a task.
// TaskSession represents a persistent agent execution session for a task.
// This replaces the in-memory TaskExecution tracking and survives backend restarts.
type AgentSession struct {
type TaskSession struct {
ID string `json:"id"`
TaskID string `json:"task_id"`
AgentInstanceID string `json:"agent_instance_id"` // Docker container/agent instance
@@ -162,7 +162,7 @@ type AgentSession struct {
ExecutorSnapshot map[string]interface{} `json:"executor_snapshot,omitempty"`
EnvironmentSnapshot map[string]interface{} `json:"environment_snapshot,omitempty"`
RepositorySnapshot map[string]interface{} `json:"repository_snapshot,omitempty"`
State AgentSessionState `json:"state"`
State TaskSessionState `json:"state"`
Progress int `json:"progress"` // 0-100
ErrorMessage string `json:"error_message,omitempty"`
Metadata map[string]interface{} `json:"metadata,omitempty"`
@@ -171,9 +171,9 @@ type AgentSession struct {
UpdatedAt time.Time `json:"updated_at"`
}
// ToAPI converts internal AgentSession to API type
// TODO: Add v1.AgentSession type to pkg/api/v1/
func (s *AgentSession) ToAPI() map[string]interface{} {
// ToAPI converts internal TaskSession to API type
// TODO: Add v1.TaskSession type to pkg/api/v1/
func (s *TaskSession) ToAPI() map[string]interface{} {
result := map[string]interface{}{
"id": s.ID,
"task_id": s.TaskID,
apps/backend/internal/task/repository/interface.go
+14 -14
View File
@@ -58,20 +58,20 @@ type Repository interface {
ListMessagesPaginated(ctx context.Context, sessionID string, opts ListMessagesOptions) ([]*models.Message, bool, error)
DeleteMessage(ctx context.Context, id string) error
// Agent Session operations
CreateAgentSession(ctx context.Context, session *models.AgentSession) error
GetAgentSession(ctx context.Context, id string) (*models.AgentSession, error)
GetAgentSessionByTaskID(ctx context.Context, taskID string) (*models.AgentSession, error)
GetActiveAgentSessionByTaskID(ctx context.Context, taskID string) (*models.AgentSession, error)
UpdateAgentSession(ctx context.Context, session *models.AgentSession) error
UpdateAgentSessionState(ctx context.Context, id string, state models.AgentSessionState, errorMessage string) error
ListAgentSessions(ctx context.Context, taskID string) ([]*models.AgentSession, error)
ListActiveAgentSessions(ctx context.Context) ([]*models.AgentSession, error)
HasActiveAgentSessionsByAgentProfile(ctx context.Context, agentProfileID string) (bool, error)
HasActiveAgentSessionsByExecutor(ctx context.Context, executorID string) (bool, error)
HasActiveAgentSessionsByEnvironment(ctx context.Context, environmentID string) (bool, error)
HasActiveAgentSessionsByRepository(ctx context.Context, repositoryID string) (bool, error)
DeleteAgentSession(ctx context.Context, id string) error
// Task Session operations
CreateTaskSession(ctx context.Context, session *models.TaskSession) error
GetTaskSession(ctx context.Context, id string) (*models.TaskSession, error)
GetTaskSessionByTaskID(ctx context.Context, taskID string) (*models.TaskSession, error)
GetActiveTaskSessionByTaskID(ctx context.Context, taskID string) (*models.TaskSession, error)
UpdateTaskSession(ctx context.Context, session *models.TaskSession) error
UpdateTaskSessionState(ctx context.Context, id string, state models.TaskSessionState, errorMessage string) error
ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error)
ListActiveTaskSessions(ctx context.Context) ([]*models.TaskSession, error)
HasActiveTaskSessionsByAgentProfile(ctx context.Context, agentProfileID string) (bool, error)
HasActiveTaskSessionsByExecutor(ctx context.Context, executorID string) (bool, error)
HasActiveTaskSessionsByEnvironment(ctx context.Context, environmentID string) (bool, error)
HasActiveTaskSessionsByRepository(ctx context.Context, repositoryID string) (bool, error)
DeleteTaskSession(ctx context.Context, id string) error
// Repository operations
CreateRepository(ctx context.Context, repository *models.Repository) error
apps/backend/internal/task/repository/memory.go
+57 -57
View File
@@ -22,7 +22,7 @@ type MemoryRepository struct {
messages map[string]*models.Message
repositories map[string]*models.Repository
repoScripts map[string]*models.RepositoryScript
agentSessions map[string]*models.AgentSession
taskSessions map[string]*models.TaskSession
executors map[string]*models.Executor
environments map[string]*models.Environment
taskBoards map[string]map[string]struct{}
@@ -43,7 +43,7 @@ func NewMemoryRepository() *MemoryRepository {
messages: make(map[string]*models.Message),
repositories: make(map[string]*models.Repository),
repoScripts: make(map[string]*models.RepositoryScript),
agentSessions: make(map[string]*models.AgentSession),
taskSessions: make(map[string]*models.TaskSession),
executors: make(map[string]*models.Executor),
environments: make(map[string]*models.Environment),
taskBoards: make(map[string]map[string]struct{}),
@@ -619,11 +619,11 @@ func (r *MemoryRepository) addTaskPlacement(taskID, boardID, columnID string, po
}
}
func isSessionActive(status models.AgentSessionState) bool {
return status == models.AgentSessionStateCreated ||
status == models.AgentSessionStateRunning ||
status == models.AgentSessionStateWaitingForInput ||
status == models.AgentSessionStateStarting
func isSessionActive(status models.TaskSessionState) bool {
return status == models.TaskSessionStateCreated ||
status == models.TaskSessionStateRunning ||
status == models.TaskSessionStateWaitingForInput ||
status == models.TaskSessionStateStarting
}
// Message operations
@@ -668,7 +668,7 @@ func (r *MemoryRepository) GetMessageByToolCallID(ctx context.Context, sessionID
defer r.mu.RUnlock()
for _, message := range r.messages {
if message.AgentSessionID != sessionID || message.Metadata == nil {
if message.TaskSessionID != sessionID || message.Metadata == nil {
continue
}
if value, ok := message.Metadata["tool_call_id"]; ok {
@@ -701,7 +701,7 @@ func (r *MemoryRepository) ListMessages(ctx context.Context, sessionID string) (
var result []*models.Message
for _, message := range r.messages {
if message.AgentSessionID == sessionID {
if message.TaskSessionID == sessionID {
result = append(result, message)
}
}
@@ -730,7 +730,7 @@ func (r *MemoryRepository) ListMessagesPaginated(ctx context.Context, sessionID
if !ok {
return nil, false, fmt.Errorf("message not found: %s", opts.Before)
}
if message.AgentSessionID != sessionID {
if message.TaskSessionID != sessionID {
return nil, false, fmt.Errorf("message cursor not found: %s", opts.Before)
}
cursor = message
@@ -740,7 +740,7 @@ func (r *MemoryRepository) ListMessagesPaginated(ctx context.Context, sessionID
if !ok {
return nil, false, fmt.Errorf("message not found: %s", opts.After)
}
if message.AgentSessionID != sessionID {
if message.TaskSessionID != sessionID {
return nil, false, fmt.Errorf("message cursor not found: %s", opts.After)
}
cursor = message
@@ -748,7 +748,7 @@ func (r *MemoryRepository) ListMessagesPaginated(ctx context.Context, sessionID
var result []*models.Message
for _, message := range r.messages {
if message.AgentSessionID == sessionID {
if message.TaskSessionID == sessionID {
result = append(result, message)
}
}
@@ -801,10 +801,10 @@ func (r *MemoryRepository) DeleteMessage(ctx context.Context, id string) error {
return nil
}
// AgentSession operations
// TaskSession operations
// CreateAgentSession creates a new agent session
func (r *MemoryRepository) CreateAgentSession(ctx context.Context, session *models.AgentSession) error {
// CreateTaskSession creates a new agent session
func (r *MemoryRepository) CreateTaskSession(ctx context.Context, session *models.TaskSession) error {
r.mu.Lock()
defer r.mu.Unlock()
@@ -816,33 +816,33 @@ func (r *MemoryRepository) CreateAgentSession(ctx context.Context, session *mode
session.StartedAt = now
}
if session.State == "" {
session.State = models.AgentSessionStateCreated
session.State = models.TaskSessionStateCreated
}
session.UpdatedAt = now
r.agentSessions[session.ID] = session
r.taskSessions[session.ID] = session
return nil
}
// GetAgentSession retrieves an agent session by ID
func (r *MemoryRepository) GetAgentSession(ctx context.Context, id string) (*models.AgentSession, error) {
// GetTaskSession retrieves an agent session by ID
func (r *MemoryRepository) GetTaskSession(ctx context.Context, id string) (*models.TaskSession, error) {
r.mu.RLock()
defer r.mu.RUnlock()
session, ok := r.agentSessions[id]
session, ok := r.taskSessions[id]
if !ok {
return nil, fmt.Errorf("agent session not found: %s", id)
}
return session, nil
}
// GetAgentSessionByTaskID retrieves the most recent agent session for a task
func (r *MemoryRepository) GetAgentSessionByTaskID(ctx context.Context, taskID string) (*models.AgentSession, error) {
// GetTaskSessionByTaskID retrieves the most recent agent session for a task
func (r *MemoryRepository) GetTaskSessionByTaskID(ctx context.Context, taskID string) (*models.TaskSession, error) {
r.mu.RLock()
defer r.mu.RUnlock()
var latest *models.AgentSession
for _, session := range r.agentSessions {
var latest *models.TaskSession
for _, session := range r.taskSessions {
if session.TaskID == taskID {
if latest == nil || session.StartedAt.After(latest.StartedAt) {
latest = session
@@ -855,58 +855,58 @@ func (r *MemoryRepository) GetAgentSessionByTaskID(ctx context.Context, taskID s
return latest, nil
}
// GetActiveAgentSessionByTaskID retrieves the active agent session for a task
func (r *MemoryRepository) GetActiveAgentSessionByTaskID(ctx context.Context, taskID string) (*models.AgentSession, error) {
// GetActiveTaskSessionByTaskID retrieves the active agent session for a task
func (r *MemoryRepository) GetActiveTaskSessionByTaskID(ctx context.Context, taskID string) (*models.TaskSession, error) {
r.mu.RLock()
defer r.mu.RUnlock()
for _, session := range r.agentSessions {
if session.TaskID == taskID && (session.State == models.AgentSessionStateCreated || session.State == models.AgentSessionStateStarting || session.State == models.AgentSessionStateRunning || session.State == models.AgentSessionStateWaitingForInput) {
for _, session := range r.taskSessions {
if session.TaskID == taskID && (session.State == models.TaskSessionStateCreated || session.State == models.TaskSessionStateStarting || session.State == models.TaskSessionStateRunning || session.State == models.TaskSessionStateWaitingForInput) {
return session, nil
}
}
return nil, fmt.Errorf("no active agent session found for task: %s", taskID)
}
// UpdateAgentSession updates an existing agent session
func (r *MemoryRepository) UpdateAgentSession(ctx context.Context, session *models.AgentSession) error {
// UpdateTaskSession updates an existing agent session
func (r *MemoryRepository) UpdateTaskSession(ctx context.Context, session *models.TaskSession) error {
r.mu.Lock()
defer r.mu.Unlock()
if _, ok := r.agentSessions[session.ID]; !ok {
if _, ok := r.taskSessions[session.ID]; !ok {
return fmt.Errorf("agent session not found: %s", session.ID)
}
session.UpdatedAt = time.Now().UTC()
r.agentSessions[session.ID] = session
r.taskSessions[session.ID] = session
return nil
}
// UpdateAgentSessionState updates the status of an agent session
func (r *MemoryRepository) UpdateAgentSessionState(ctx context.Context, id string, status models.AgentSessionState, errorMessage string) error {
// UpdateTaskSessionState updates the status of an agent session
func (r *MemoryRepository) UpdateTaskSessionState(ctx context.Context, id string, status models.TaskSessionState, errorMessage string) error {
r.mu.Lock()
defer r.mu.Unlock()
session, ok := r.agentSessions[id]
session, ok := r.taskSessions[id]
if !ok {
return fmt.Errorf("agent session not found: %s", id)
}
session.State = status
session.ErrorMessage = errorMessage
session.UpdatedAt = time.Now().UTC()
if status == models.AgentSessionStateCompleted || status == models.AgentSessionStateFailed || status == models.AgentSessionStateCancelled {
if status == models.TaskSessionStateCompleted || status == models.TaskSessionStateFailed || status == models.TaskSessionStateCancelled {
now := time.Now().UTC()
session.CompletedAt = &now
}
return nil
}
// ListAgentSessions returns all agent sessions for a task
func (r *MemoryRepository) ListAgentSessions(ctx context.Context, taskID string) ([]*models.AgentSession, error) {
// ListTaskSessions returns all agent sessions for a task
func (r *MemoryRepository) ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error) {
r.mu.RLock()
defer r.mu.RUnlock()
var result []*models.AgentSession
for _, session := range r.agentSessions {
var result []*models.TaskSession
for _, session := range r.taskSessions {
if session.TaskID == taskID {
result = append(result, session)
}
@@ -914,25 +914,25 @@ func (r *MemoryRepository) ListAgentSessions(ctx context.Context, taskID string)
return result, nil
}
// ListActiveAgentSessions returns all active agent sessions
func (r *MemoryRepository) ListActiveAgentSessions(ctx context.Context) ([]*models.AgentSession, error) {
// ListActiveTaskSessions returns all active agent sessions
func (r *MemoryRepository) ListActiveTaskSessions(ctx context.Context) ([]*models.TaskSession, error) {
r.mu.RLock()
defer r.mu.RUnlock()
var result []*models.AgentSession
for _, session := range r.agentSessions {
if session.State == models.AgentSessionStateCreated || session.State == models.AgentSessionStateStarting || session.State == models.AgentSessionStateRunning || session.State == models.AgentSessionStateWaitingForInput {
var result []*models.TaskSession
for _, session := range r.taskSessions {
if session.State == models.TaskSessionStateCreated || session.State == models.TaskSessionStateStarting || session.State == models.TaskSessionStateRunning || session.State == models.TaskSessionStateWaitingForInput {
result = append(result, session)
}
}
return result, nil
}
func (r *MemoryRepository) HasActiveAgentSessionsByAgentProfile(ctx context.Context, agentProfileID string) (bool, error) {
func (r *MemoryRepository) HasActiveTaskSessionsByAgentProfile(ctx context.Context, agentProfileID string) (bool, error) {
r.mu.RLock()
defer r.mu.RUnlock()
for _, session := range r.agentSessions {
for _, session := range r.taskSessions {
if session.AgentProfileID == agentProfileID && isSessionActive(session.State) {
return true, nil
}
@@ -940,11 +940,11 @@ func (r *MemoryRepository) HasActiveAgentSessionsByAgentProfile(ctx context.Cont
return false, nil
}
func (r *MemoryRepository) HasActiveAgentSessionsByExecutor(ctx context.Context, executorID string) (bool, error) {
func (r *MemoryRepository) HasActiveTaskSessionsByExecutor(ctx context.Context, executorID string) (bool, error) {
r.mu.RLock()
defer r.mu.RUnlock()
for _, session := range r.agentSessions {
for _, session := range r.taskSessions {
if session.ExecutorID == executorID && isSessionActive(session.State) {
return true, nil
}
@@ -952,11 +952,11 @@ func (r *MemoryRepository) HasActiveAgentSessionsByExecutor(ctx context.Context,
return false, nil
}
func (r *MemoryRepository) HasActiveAgentSessionsByEnvironment(ctx context.Context, environmentID string) (bool, error) {
func (r *MemoryRepository) HasActiveTaskSessionsByEnvironment(ctx context.Context, environmentID string) (bool, error) {
r.mu.RLock()
defer r.mu.RUnlock()
for _, session := range r.agentSessions {
for _, session := range r.taskSessions {
if session.EnvironmentID == environmentID && isSessionActive(session.State) {
return true, nil
}
@@ -964,11 +964,11 @@ func (r *MemoryRepository) HasActiveAgentSessionsByEnvironment(ctx context.Conte
return false, nil
}
func (r *MemoryRepository) HasActiveAgentSessionsByRepository(ctx context.Context, repositoryID string) (bool, error) {
func (r *MemoryRepository) HasActiveTaskSessionsByRepository(ctx context.Context, repositoryID string) (bool, error) {
r.mu.RLock()
defer r.mu.RUnlock()
for _, session := range r.agentSessions {
for _, session := range r.taskSessions {
if !isSessionActive(session.State) {
continue
}
@@ -983,15 +983,15 @@ func (r *MemoryRepository) HasActiveAgentSessionsByRepository(ctx context.Contex
return false, nil
}
// DeleteAgentSession deletes an agent session by ID
func (r *MemoryRepository) DeleteAgentSession(ctx context.Context, id string) error {
// DeleteTaskSession deletes an agent session by ID
func (r *MemoryRepository) DeleteTaskSession(ctx context.Context, id string) error {
r.mu.Lock()
defer r.mu.Unlock()
if _, ok := r.agentSessions[id]; !ok {
if _, ok := r.taskSessions[id]; !ok {
return fmt.Errorf("agent session not found: %s", id)
}
delete(r.agentSessions, id)
delete(r.taskSessions, id)
return nil
}
apps/backend/internal/task/repository/memory_test.go
+11 -11
View File
@@ -351,13 +351,13 @@ func TestMemoryRepository_ListTasksByColumn(t *testing.T) {
// Message CRUD tests
func setupMemoryTestSession(ctx context.Context, repo *MemoryRepository, taskID string) string {
session := &models.AgentSession{
session := &models.TaskSession{
ID: "session-123",
TaskID: taskID,
AgentProfileID: "profile-123",
State: models.AgentSessionStateStarting,
State: models.TaskSessionStateStarting,
}
_ = repo.CreateAgentSession(ctx, session)
_ = repo.CreateTaskSession(ctx, session)
return session.ID
}
@@ -372,7 +372,7 @@ func TestMemoryRepository_MessageCRUD(t *testing.T) {
// Create comment
comment := &models.Message{
AgentSessionID: sessionID,
TaskSessionID: sessionID,
TaskID: "task-123",
AuthorType: models.MessageAuthorUser,
AuthorID: "user-123",
@@ -436,9 +436,9 @@ func TestMemoryRepository_ListMessages(t *testing.T) {
sessionID := setupMemoryTestSession(ctx, repo, task.ID)
// Create multiple comments
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-1", AgentSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorUser, Content: "Comment 1"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-2", AgentSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorAgent, Content: "Comment 2"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-3", AgentSessionID: "session-456", TaskID: "task-456", AuthorType: models.MessageAuthorUser, Content: "Comment 3"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-1", TaskSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorUser, Content: "Comment 1"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-2", TaskSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorAgent, Content: "Comment 2"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-3", TaskSessionID: "session-456", TaskID: "task-456", AuthorType: models.MessageAuthorUser, Content: "Comment 3"})
comments, err := repo.ListMessages(ctx, sessionID)
if err != nil {
@@ -460,7 +460,7 @@ func TestMemoryRepository_ListMessagesPagination(t *testing.T) {
baseTime := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
_ = repo.CreateMessage(ctx, &models.Message{
ID: "comment-1",
AgentSessionID: sessionID,
TaskSessionID: sessionID,
TaskID: "task-123",
AuthorType: models.MessageAuthorUser,
Content: "Comment 1",
@@ -468,7 +468,7 @@ func TestMemoryRepository_ListMessagesPagination(t *testing.T) {
})
_ = repo.CreateMessage(ctx, &models.Message{
ID: "comment-2",
AgentSessionID: sessionID,
TaskSessionID: sessionID,
TaskID: "task-123",
AuthorType: models.MessageAuthorUser,
Content: "Comment 2",
@@ -476,7 +476,7 @@ func TestMemoryRepository_ListMessagesPagination(t *testing.T) {
})
_ = repo.CreateMessage(ctx, &models.Message{
ID: "comment-3",
AgentSessionID: sessionID,
TaskSessionID: sessionID,
TaskID: "task-123",
AuthorType: models.MessageAuthorUser,
Content: "Comment 3",
@@ -512,7 +512,7 @@ func TestMemoryRepository_MessageWithRequestsInput(t *testing.T) {
// Create agent comment requesting input
comment := &models.Message{
AgentSessionID: sessionID,
TaskSessionID: sessionID,
TaskID: "task-123",
AuthorType: models.MessageAuthorAgent,
AuthorID: "agent-123",
apps/backend/internal/task/repository/sqlite.go
+68 -68
View File
@@ -204,7 +204,7 @@ func (r *SQLiteRepository) initSchema() error {
CREATE TABLE IF NOT EXISTS task_messages (
id TEXT PRIMARY KEY,
agent_session_id TEXT NOT NULL,
task_session_id TEXT NOT NULL,
task_id TEXT DEFAULT '',
author_type TEXT NOT NULL DEFAULT 'user',
author_id TEXT DEFAULT '',
@@ -213,14 +213,14 @@ func (r *SQLiteRepository) initSchema() error {
type TEXT NOT NULL DEFAULT 'message',
metadata TEXT DEFAULT '{}',
created_at DATETIME NOT NULL,
FOREIGN KEY (agent_session_id) REFERENCES agent_sessions(id) ON DELETE CASCADE
FOREIGN KEY (task_session_id) REFERENCES task_sessions(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_messages_session_id ON task_messages(agent_session_id);
CREATE INDEX IF NOT EXISTS idx_messages_session_id ON task_messages(task_session_id);
CREATE INDEX IF NOT EXISTS idx_messages_created_at ON task_messages(created_at);
CREATE INDEX IF NOT EXISTS idx_messages_session_created ON task_messages(agent_session_id, created_at);
CREATE INDEX IF NOT EXISTS idx_messages_session_created ON task_messages(task_session_id, created_at);
CREATE TABLE IF NOT EXISTS agent_sessions (
CREATE TABLE IF NOT EXISTS task_sessions (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL,
agent_instance_id TEXT NOT NULL DEFAULT '',
@@ -247,9 +247,9 @@ func (r *SQLiteRepository) initSchema() error {
FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_agent_sessions_task_id ON agent_sessions(task_id);
CREATE INDEX IF NOT EXISTS idx_agent_sessions_state ON agent_sessions(state);
CREATE INDEX IF NOT EXISTS idx_agent_sessions_task_state ON agent_sessions(task_id, state);
CREATE INDEX IF NOT EXISTS idx_task_sessions_task_id ON task_sessions(task_id);
CREATE INDEX IF NOT EXISTS idx_task_sessions_state ON task_sessions(state);
CREATE INDEX IF NOT EXISTS idx_task_sessions_task_state ON task_sessions(task_id, state);
`
if _, err := r.db.Exec(schema); err != nil {
@@ -294,21 +294,21 @@ func (r *SQLiteRepository) initSchema() error {
if err := r.ensureColumn("task_messages", "metadata", "TEXT DEFAULT '{}'"); err != nil {
return err
}
if err := r.ensureColumn("task_messages", "agent_session_id", "TEXT NOT NULL DEFAULT ''"); err != nil {
if err := r.ensureColumn("task_messages", "task_session_id", "TEXT NOT NULL DEFAULT ''"); err != nil {
return err
}
// Ensure container_id column exists for existing databases
if err := r.ensureColumn("agent_sessions", "container_id", "TEXT NOT NULL DEFAULT ''"); err != nil {
if err := r.ensureColumn("task_sessions", "container_id", "TEXT NOT NULL DEFAULT ''"); err != nil {
return err
}
if err := r.ensureColumn("agent_sessions", "executor_id", "TEXT DEFAULT ''"); err != nil {
if err := r.ensureColumn("task_sessions", "executor_id", "TEXT DEFAULT ''"); err != nil {
return err
}
if err := r.ensureColumn("agent_sessions", "environment_id", "TEXT DEFAULT ''"); err != nil {
if err := r.ensureColumn("task_sessions", "environment_id", "TEXT DEFAULT ''"); err != nil {
return err
}
if err := r.ensureColumn("agent_sessions", "state", "TEXT NOT NULL DEFAULT 'CREATED'"); err != nil {
if err := r.ensureColumn("task_sessions", "state", "TEXT NOT NULL DEFAULT 'CREATED'"); err != nil {
return err
}
@@ -1178,9 +1178,9 @@ func (r *SQLiteRepository) CreateMessage(ctx context.Context, message *models.Me
}
_, err := r.db.ExecContext(ctx, `
INSERT INTO task_messages (id, agent_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at)
INSERT INTO task_messages (id, task_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, message.ID, message.AgentSessionID, message.TaskID, message.AuthorType, message.AuthorID, message.Content, requestsInput, messageType, metadataJSON, message.CreatedAt)
`, message.ID, message.TaskSessionID, message.TaskID, message.AuthorType, message.AuthorID, message.Content, requestsInput, messageType, metadataJSON, message.CreatedAt)
return err
}
@@ -1192,9 +1192,9 @@ func (r *SQLiteRepository) GetMessage(ctx context.Context, id string) (*models.M
var messageType string
var metadataJSON string
err := r.db.QueryRowContext(ctx, `
SELECT id, agent_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at
SELECT id, task_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at
FROM task_messages WHERE id = ?
`, id).Scan(&message.ID, &message.AgentSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
`, id).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
if err != nil {
return nil, err
}
@@ -1213,8 +1213,8 @@ func (r *SQLiteRepository) GetMessage(ctx context.Context, id string) (*models.M
// ListMessages returns all messages for a session ordered by creation time.
func (r *SQLiteRepository) ListMessages(ctx context.Context, sessionID string) ([]*models.Message, error) {
rows, err := r.db.QueryContext(ctx, `
SELECT id, agent_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at
FROM task_messages WHERE agent_session_id = ? ORDER BY created_at ASC
SELECT id, task_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at
FROM task_messages WHERE task_session_id = ? ORDER BY created_at ASC
`, sessionID)
if err != nil {
return nil, err
@@ -1227,7 +1227,7 @@ func (r *SQLiteRepository) ListMessages(ctx context.Context, sessionID string) (
var requestsInput int
var messageType string
var metadataJSON string
err := rows.Scan(&message.ID, &message.AgentSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
err := rows.Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
if err != nil {
return nil, err
}
@@ -1267,7 +1267,7 @@ func (r *SQLiteRepository) ListMessagesPaginated(ctx context.Context, sessionID
if err != nil {
return nil, false, err
}
if cursor.AgentSessionID != sessionID {
if cursor.TaskSessionID != sessionID {
return nil, false, fmt.Errorf("message cursor not found: %s", opts.Before)
}
}
@@ -1277,14 +1277,14 @@ func (r *SQLiteRepository) ListMessagesPaginated(ctx context.Context, sessionID
if err != nil {
return nil, false, err
}
if cursor.AgentSessionID != sessionID {
if cursor.TaskSessionID != sessionID {
return nil, false, fmt.Errorf("message cursor not found: %s", opts.After)
}
}
query := `
SELECT id, agent_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at
FROM task_messages WHERE agent_session_id = ?`
SELECT id, task_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at
FROM task_messages WHERE task_session_id = ?`
args := []interface{}{sessionID}
if cursor != nil {
if opts.Before != "" {
@@ -1312,7 +1312,7 @@ func (r *SQLiteRepository) ListMessagesPaginated(ctx context.Context, sessionID
var requestsInput int
var messageType string
var metadataJSON string
err := rows.Scan(&message.ID, &message.AgentSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
err := rows.Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
if err != nil {
return nil, false, err
}
@@ -1359,9 +1359,9 @@ func (r *SQLiteRepository) GetMessageByToolCallID(ctx context.Context, sessionID
var messageType string
var metadataJSON string
err := r.db.QueryRowContext(ctx, `
SELECT id, agent_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at
FROM task_messages WHERE agent_session_id = ? AND json_extract(metadata, '$.tool_call_id') = ?
`, sessionID, toolCallID).Scan(&message.ID, &message.AgentSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID,
SELECT id, task_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at
FROM task_messages WHERE task_session_id = ? AND json_extract(metadata, '$.tool_call_id') = ?
`, sessionID, toolCallID).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID,
&message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
if err != nil {
return nil, err
@@ -1598,7 +1598,7 @@ func (r *SQLiteRepository) ListRepositoryScripts(ctx context.Context, repository
// Agent Session operations
// CreateAgentSession creates a new agent session
func (r *SQLiteRepository) CreateAgentSession(ctx context.Context, session *models.AgentSession) error {
func (r *SQLiteRepository) CreateTaskSession(ctx context.Context, session *models.TaskSession) error {
if session.ID == "" {
session.ID = uuid.New().String()
}
@@ -1606,7 +1606,7 @@ func (r *SQLiteRepository) CreateAgentSession(ctx context.Context, session *mode
session.StartedAt = now
session.UpdatedAt = now
if session.State == "" {
session.State = models.AgentSessionStateCreated
session.State = models.TaskSessionStateCreated
}
metadataJSON, err := json.Marshal(session.Metadata)
@@ -1631,7 +1631,7 @@ func (r *SQLiteRepository) CreateAgentSession(ctx context.Context, session *mode
}
_, err = r.db.ExecContext(ctx, `
INSERT INTO agent_sessions (
INSERT INTO task_sessions (
id, task_id, agent_instance_id, container_id, agent_profile_id, executor_id, environment_id,
repository_id, base_branch, worktree_id, worktree_path, worktree_branch,
agent_profile_snapshot, executor_snapshot, environment_snapshot, repository_snapshot,
@@ -1647,8 +1647,8 @@ func (r *SQLiteRepository) CreateAgentSession(ctx context.Context, session *mode
}
// GetAgentSession retrieves an agent session by ID
func (r *SQLiteRepository) GetAgentSession(ctx context.Context, id string) (*models.AgentSession, error) {
session := &models.AgentSession{}
func (r *SQLiteRepository) GetTaskSession(ctx context.Context, id string) (*models.TaskSession, error) {
session := &models.TaskSession{}
var state string
var metadataJSON string
var agentProfileSnapshotJSON string
@@ -1662,7 +1662,7 @@ func (r *SQLiteRepository) GetAgentSession(ctx context.Context, id string) (*mod
repository_id, base_branch, worktree_id, worktree_path, worktree_branch,
agent_profile_snapshot, executor_snapshot, environment_snapshot, repository_snapshot,
state, progress, error_message, metadata, started_at, completed_at, updated_at
FROM agent_sessions WHERE id = ?
FROM task_sessions WHERE id = ?
`, id).Scan(
&session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentProfileID,
&session.ExecutorID, &session.EnvironmentID,
@@ -1678,7 +1678,7 @@ func (r *SQLiteRepository) GetAgentSession(ctx context.Context, id string) (*mod
return nil, err
}
session.State = models.AgentSessionState(state)
session.State = models.TaskSessionState(state)
if completedAt.Valid {
session.CompletedAt = &completedAt.Time
}
@@ -1712,8 +1712,8 @@ func (r *SQLiteRepository) GetAgentSession(ctx context.Context, id string) (*mod
}
// GetAgentSessionByTaskID retrieves the most recent agent session for a task
func (r *SQLiteRepository) GetAgentSessionByTaskID(ctx context.Context, taskID string) (*models.AgentSession, error) {
session := &models.AgentSession{}
func (r *SQLiteRepository) GetTaskSessionByTaskID(ctx context.Context, taskID string) (*models.TaskSession, error) {
session := &models.TaskSession{}
var state string
var metadataJSON string
var agentProfileSnapshotJSON string
@@ -1727,7 +1727,7 @@ func (r *SQLiteRepository) GetAgentSessionByTaskID(ctx context.Context, taskID s
repository_id, base_branch, worktree_id, worktree_path, worktree_branch,
agent_profile_snapshot, executor_snapshot, environment_snapshot, repository_snapshot,
state, progress, error_message, metadata, started_at, completed_at, updated_at
FROM agent_sessions WHERE task_id = ? ORDER BY started_at DESC LIMIT 1
FROM task_sessions WHERE task_id = ? ORDER BY started_at DESC LIMIT 1
`, taskID).Scan(
&session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentProfileID,
&session.ExecutorID, &session.EnvironmentID,
@@ -1743,7 +1743,7 @@ func (r *SQLiteRepository) GetAgentSessionByTaskID(ctx context.Context, taskID s
return nil, err
}
session.State = models.AgentSessionState(state)
session.State = models.TaskSessionState(state)
if completedAt.Valid {
session.CompletedAt = &completedAt.Time
}
@@ -1777,8 +1777,8 @@ func (r *SQLiteRepository) GetAgentSessionByTaskID(ctx context.Context, taskID s
}
// GetActiveAgentSessionByTaskID retrieves the active (running/waiting) agent session for a task
func (r *SQLiteRepository) GetActiveAgentSessionByTaskID(ctx context.Context, taskID string) (*models.AgentSession, error) {
session := &models.AgentSession{}
func (r *SQLiteRepository) GetActiveTaskSessionByTaskID(ctx context.Context, taskID string) (*models.TaskSession, error) {
session := &models.TaskSession{}
var state string
var metadataJSON string
var agentProfileSnapshotJSON string
@@ -1792,7 +1792,7 @@ func (r *SQLiteRepository) GetActiveAgentSessionByTaskID(ctx context.Context, ta
repository_id, base_branch, worktree_id, worktree_path, worktree_branch,
agent_profile_snapshot, executor_snapshot, environment_snapshot, repository_snapshot,
state, progress, error_message, metadata, started_at, completed_at, updated_at
FROM agent_sessions
FROM task_sessions
WHERE task_id = ? AND state IN ('CREATED', 'STARTING', 'RUNNING', 'WAITING_FOR_INPUT')
ORDER BY started_at DESC LIMIT 1
`, taskID).Scan(
@@ -1810,7 +1810,7 @@ func (r *SQLiteRepository) GetActiveAgentSessionByTaskID(ctx context.Context, ta
return nil, err
}
session.State = models.AgentSessionState(state)
session.State = models.TaskSessionState(state)
if completedAt.Valid {
session.CompletedAt = &completedAt.Time
}
@@ -1844,7 +1844,7 @@ func (r *SQLiteRepository) GetActiveAgentSessionByTaskID(ctx context.Context, ta
}
// UpdateAgentSession updates an existing agent session
func (r *SQLiteRepository) UpdateAgentSession(ctx context.Context, session *models.AgentSession) error {
func (r *SQLiteRepository) UpdateTaskSession(ctx context.Context, session *models.TaskSession) error {
session.UpdatedAt = time.Now().UTC()
metadataJSON, err := json.Marshal(session.Metadata)
@@ -1869,7 +1869,7 @@ func (r *SQLiteRepository) UpdateAgentSession(ctx context.Context, session *mode
}
result, err := r.db.ExecContext(ctx, `
UPDATE agent_sessions SET
UPDATE task_sessions SET
agent_instance_id = ?, container_id = ?, agent_profile_id = ?, executor_id = ?, environment_id = ?,
repository_id = ?, base_branch = ?, worktree_id = ?, worktree_path = ?, worktree_branch = ?,
agent_profile_snapshot = ?, executor_snapshot = ?, environment_snapshot = ?, repository_snapshot = ?,
@@ -1892,16 +1892,16 @@ func (r *SQLiteRepository) UpdateAgentSession(ctx context.Context, session *mode
}
// UpdateAgentSessionState updates just the state and error message of an agent session
func (r *SQLiteRepository) UpdateAgentSessionState(ctx context.Context, id string, status models.AgentSessionState, errorMessage string) error {
func (r *SQLiteRepository) UpdateTaskSessionState(ctx context.Context, id string, status models.TaskSessionState, errorMessage string) error {
now := time.Now().UTC()
var completedAt *time.Time
if status == models.AgentSessionStateCompleted || status == models.AgentSessionStateFailed || status == models.AgentSessionStateCancelled {
if status == models.TaskSessionStateCompleted || status == models.TaskSessionStateFailed || status == models.TaskSessionStateCancelled {
completedAt = &now
}
result, err := r.db.ExecContext(ctx, `
UPDATE agent_sessions SET state = ?, error_message = ?, completed_at = ?, updated_at = ? WHERE id = ?
UPDATE task_sessions SET state = ?, error_message = ?, completed_at = ?, updated_at = ? WHERE id = ?
`, string(status), errorMessage, completedAt, now, id)
if err != nil {
return err
@@ -1915,43 +1915,43 @@ func (r *SQLiteRepository) UpdateAgentSessionState(ctx context.Context, id strin
}
// ListAgentSessions returns all agent sessions for a task
func (r *SQLiteRepository) ListAgentSessions(ctx context.Context, taskID string) ([]*models.AgentSession, error) {
func (r *SQLiteRepository) ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error) {
rows, err := r.db.QueryContext(ctx, `
SELECT id, task_id, agent_instance_id, container_id, agent_profile_id, executor_id, environment_id,
repository_id, base_branch, worktree_id, worktree_path, worktree_branch,
agent_profile_snapshot, executor_snapshot, environment_snapshot, repository_snapshot,
state, progress, error_message, metadata, started_at, completed_at, updated_at
FROM agent_sessions WHERE task_id = ? ORDER BY started_at DESC
FROM task_sessions WHERE task_id = ? ORDER BY started_at DESC
`, taskID)
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
return r.scanAgentSessions(rows)
return r.scanTaskSessions(rows)
}
// ListActiveAgentSessions returns all active agent sessions across all tasks
func (r *SQLiteRepository) ListActiveAgentSessions(ctx context.Context) ([]*models.AgentSession, error) {
func (r *SQLiteRepository) ListActiveTaskSessions(ctx context.Context) ([]*models.TaskSession, error) {
rows, err := r.db.QueryContext(ctx, `
SELECT id, task_id, agent_instance_id, container_id, agent_profile_id, executor_id, environment_id,
repository_id, base_branch, worktree_id, worktree_path, worktree_branch,
agent_profile_snapshot, executor_snapshot, environment_snapshot, repository_snapshot,
state, progress, error_message, metadata, started_at, completed_at, updated_at
FROM agent_sessions WHERE state IN ('CREATED', 'STARTING', 'RUNNING', 'WAITING_FOR_INPUT') ORDER BY started_at DESC
FROM task_sessions WHERE state IN ('CREATED', 'STARTING', 'RUNNING', 'WAITING_FOR_INPUT') ORDER BY started_at DESC
`)
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
return r.scanAgentSessions(rows)
return r.scanTaskSessions(rows)
}
func (r *SQLiteRepository) HasActiveAgentSessionsByAgentProfile(ctx context.Context, agentProfileID string) (bool, error) {
func (r *SQLiteRepository) HasActiveTaskSessionsByAgentProfile(ctx context.Context, agentProfileID string) (bool, error) {
var exists int
err := r.db.QueryRowContext(ctx, `
SELECT 1 FROM agent_sessions
SELECT 1 FROM task_sessions
WHERE agent_profile_id = ? AND state IN ('CREATED', 'STARTING', 'RUNNING', 'WAITING_FOR_INPUT')
LIMIT 1
`, agentProfileID).Scan(&exists)
@@ -1961,10 +1961,10 @@ func (r *SQLiteRepository) HasActiveAgentSessionsByAgentProfile(ctx context.Cont
return err == nil, err
}
func (r *SQLiteRepository) HasActiveAgentSessionsByExecutor(ctx context.Context, executorID string) (bool, error) {
func (r *SQLiteRepository) HasActiveTaskSessionsByExecutor(ctx context.Context, executorID string) (bool, error) {
var exists int
err := r.db.QueryRowContext(ctx, `
SELECT 1 FROM agent_sessions
SELECT 1 FROM task_sessions
WHERE executor_id = ? AND state IN ('CREATED', 'STARTING', 'RUNNING', 'WAITING_FOR_INPUT')
LIMIT 1
`, executorID).Scan(&exists)
@@ -1974,10 +1974,10 @@ func (r *SQLiteRepository) HasActiveAgentSessionsByExecutor(ctx context.Context,
return err == nil, err
}
func (r *SQLiteRepository) HasActiveAgentSessionsByEnvironment(ctx context.Context, environmentID string) (bool, error) {
func (r *SQLiteRepository) HasActiveTaskSessionsByEnvironment(ctx context.Context, environmentID string) (bool, error) {
var exists int
err := r.db.QueryRowContext(ctx, `
SELECT 1 FROM agent_sessions
SELECT 1 FROM task_sessions
WHERE environment_id = ? AND state IN ('CREATED', 'STARTING', 'RUNNING', 'WAITING_FOR_INPUT')
LIMIT 1
`, environmentID).Scan(&exists)
@@ -1987,11 +1987,11 @@ func (r *SQLiteRepository) HasActiveAgentSessionsByEnvironment(ctx context.Conte
return err == nil, err
}
func (r *SQLiteRepository) HasActiveAgentSessionsByRepository(ctx context.Context, repositoryID string) (bool, error) {
func (r *SQLiteRepository) HasActiveTaskSessionsByRepository(ctx context.Context, repositoryID string) (bool, error) {
var exists int
err := r.db.QueryRowContext(ctx, `
SELECT 1
FROM agent_sessions s
FROM task_sessions s
INNER JOIN tasks t ON t.id = s.task_id
WHERE s.state IN ('CREATED', 'STARTING', 'RUNNING', 'WAITING_FOR_INPUT')
AND t.repository_id = ?
@@ -2004,10 +2004,10 @@ func (r *SQLiteRepository) HasActiveAgentSessionsByRepository(ctx context.Contex
}
// scanAgentSessions is a helper to scan multiple agent session rows
func (r *SQLiteRepository) scanAgentSessions(rows *sql.Rows) ([]*models.AgentSession, error) {
var result []*models.AgentSession
func (r *SQLiteRepository) scanTaskSessions(rows *sql.Rows) ([]*models.TaskSession, error) {
var result []*models.TaskSession
for rows.Next() {
session := &models.AgentSession{}
session := &models.TaskSession{}
var state string
var metadataJSON string
var agentProfileSnapshotJSON string
@@ -2027,7 +2027,7 @@ func (r *SQLiteRepository) scanAgentSessions(rows *sql.Rows) ([]*models.AgentSes
return nil, err
}
session.State = models.AgentSessionState(state)
session.State = models.TaskSessionState(state)
if completedAt.Valid {
session.CompletedAt = &completedAt.Time
}
@@ -2063,8 +2063,8 @@ func (r *SQLiteRepository) scanAgentSessions(rows *sql.Rows) ([]*models.AgentSes
}
// DeleteAgentSession deletes an agent session by ID
func (r *SQLiteRepository) DeleteAgentSession(ctx context.Context, id string) error {
result, err := r.db.ExecContext(ctx, `DELETE FROM agent_sessions WHERE id = ?`, id)
func (r *SQLiteRepository) DeleteTaskSession(ctx context.Context, id string) error {
result, err := r.db.ExecContext(ctx, `DELETE FROM task_sessions WHERE id = ?`, id)
if err != nil {
return err
}
apps/backend/internal/task/repository/sqlite_test.go
+74 -74
View File
@@ -506,13 +506,13 @@ func TestSQLiteRepository_Persistence(t *testing.T) {
// Message CRUD tests
func setupSQLiteTestSession(ctx context.Context, repo *SQLiteRepository, taskID, sessionID string) string {
session := &models.AgentSession{
session := &models.TaskSession{
ID: sessionID,
TaskID: taskID,
AgentProfileID: "profile-123",
State: models.AgentSessionStateStarting,
State: models.TaskSessionStateStarting,
}
_ = repo.CreateAgentSession(ctx, session)
_ = repo.CreateTaskSession(ctx, session)
return session.ID
}
@@ -534,7 +534,7 @@ func TestSQLiteRepository_MessageCRUD(t *testing.T) {
// Create comment
comment := &models.Message{
AgentSessionID: sessionID,
TaskSessionID: sessionID,
TaskID: "task-123",
AuthorType: models.MessageAuthorUser,
AuthorID: "user-123",
@@ -609,9 +609,9 @@ func TestSQLiteRepository_ListMessages(t *testing.T) {
_ = setupSQLiteTestSession(ctx, repo, task2.ID, "session-456")
// Create multiple comments
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-1", AgentSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorUser, Content: "Comment 1"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-2", AgentSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorAgent, Content: "Comment 2"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-3", AgentSessionID: "session-456", TaskID: "task-456", AuthorType: models.MessageAuthorUser, Content: "Comment 3"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-1", TaskSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorUser, Content: "Comment 1"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-2", TaskSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorAgent, Content: "Comment 2"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-3", TaskSessionID: "session-456", TaskID: "task-456", AuthorType: models.MessageAuthorUser, Content: "Comment 3"})
comments, err := repo.ListMessages(ctx, sessionID)
if err != nil {
@@ -638,7 +638,7 @@ func TestSQLiteRepository_ListMessagesPagination(t *testing.T) {
baseTime := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
_ = repo.CreateMessage(ctx, &models.Message{
ID: "comment-1",
AgentSessionID: sessionID,
TaskSessionID: sessionID,
TaskID: "task-123",
AuthorType: models.MessageAuthorUser,
Content: "Comment 1",
@@ -646,7 +646,7 @@ func TestSQLiteRepository_ListMessagesPagination(t *testing.T) {
})
_ = repo.CreateMessage(ctx, &models.Message{
ID: "comment-2",
AgentSessionID: sessionID,
TaskSessionID: sessionID,
TaskID: "task-123",
AuthorType: models.MessageAuthorUser,
Content: "Comment 2",
@@ -654,7 +654,7 @@ func TestSQLiteRepository_ListMessagesPagination(t *testing.T) {
})
_ = repo.CreateMessage(ctx, &models.Message{
ID: "comment-3",
AgentSessionID: sessionID,
TaskSessionID: sessionID,
TaskID: "task-123",
AuthorType: models.MessageAuthorUser,
Content: "Comment 3",
@@ -697,7 +697,7 @@ func TestSQLiteRepository_MessageWithRequestsInput(t *testing.T) {
// Create agent comment requesting input
comment := &models.Message{
AgentSessionID: sessionID,
TaskSessionID: sessionID,
TaskID: "task-123",
AuthorType: models.MessageAuthorAgent,
AuthorID: "agent-123",
@@ -714,9 +714,9 @@ func TestSQLiteRepository_MessageWithRequestsInput(t *testing.T) {
}
}
// AgentSession CRUD tests
// TaskSession CRUD tests
func TestSQLiteRepository_AgentSessionCRUD(t *testing.T) {
func TestSQLiteRepository_TaskSessionCRUD(t *testing.T) {
repo, cleanup := createTestSQLiteRepo(t)
defer cleanup()
ctx := context.Background()
@@ -730,18 +730,18 @@ func TestSQLiteRepository_AgentSessionCRUD(t *testing.T) {
_ = repo.CreateTask(ctx, task)
// Create agent session
session := &models.AgentSession{
session := &models.TaskSession{
TaskID: "task-123",
AgentInstanceID: "instance-abc",
ContainerID: "container-xyz",
AgentProfileID: "profile-123",
ExecutorID: "executor-1",
EnvironmentID: "env-1",
State: models.AgentSessionStateStarting,
State: models.TaskSessionStateStarting,
Progress: 0,
Metadata: map[string]interface{}{"key": "value"},
}
if err := repo.CreateAgentSession(ctx, session); err != nil {
if err := repo.CreateTaskSession(ctx, session); err != nil {
t.Fatalf("failed to create agent session: %v", err)
}
if session.ID == "" {
@@ -755,7 +755,7 @@ func TestSQLiteRepository_AgentSessionCRUD(t *testing.T) {
}
// Get agent session
retrieved, err := repo.GetAgentSession(ctx, session.ID)
retrieved, err := repo.GetTaskSession(ctx, session.ID)
if err != nil {
t.Fatalf("failed to get agent session: %v", err)
}
@@ -765,7 +765,7 @@ func TestSQLiteRepository_AgentSessionCRUD(t *testing.T) {
if retrieved.AgentProfileID != "profile-123" {
t.Errorf("expected AgentProfileID 'profile-123', got %s", retrieved.AgentProfileID)
}
if retrieved.State != models.AgentSessionStateStarting {
if retrieved.State != models.TaskSessionStateStarting {
t.Errorf("expected state 'starting', got %s", retrieved.State)
}
if retrieved.Metadata["key"] != "value" {
@@ -773,13 +773,13 @@ func TestSQLiteRepository_AgentSessionCRUD(t *testing.T) {
}
// Update agent session
session.State = models.AgentSessionStateRunning
session.State = models.TaskSessionStateRunning
session.Progress = 50
if err := repo.UpdateAgentSession(ctx, session); err != nil {
if err := repo.UpdateTaskSession(ctx, session); err != nil {
t.Fatalf("failed to update agent session: %v", err)
}
retrieved, _ = repo.GetAgentSession(ctx, session.ID)
if retrieved.State != models.AgentSessionStateRunning {
retrieved, _ = repo.GetTaskSession(ctx, session.ID)
if retrieved.State != models.TaskSessionStateRunning {
t.Errorf("expected state 'running', got %s", retrieved.State)
}
if retrieved.Progress != 50 {
@@ -787,37 +787,37 @@ func TestSQLiteRepository_AgentSessionCRUD(t *testing.T) {
}
// Delete agent session
if err := repo.DeleteAgentSession(ctx, session.ID); err != nil {
if err := repo.DeleteTaskSession(ctx, session.ID); err != nil {
t.Fatalf("failed to delete agent session: %v", err)
}
_, err = repo.GetAgentSession(ctx, session.ID)
_, err = repo.GetTaskSession(ctx, session.ID)
if err == nil {
t.Error("expected agent session to be deleted")
}
}
func TestSQLiteRepository_AgentSessionNotFound(t *testing.T) {
func TestSQLiteRepository_TaskSessionNotFound(t *testing.T) {
repo, cleanup := createTestSQLiteRepo(t)
defer cleanup()
ctx := context.Background()
_, err := repo.GetAgentSession(ctx, "nonexistent")
_, err := repo.GetTaskSession(ctx, "nonexistent")
if err == nil {
t.Error("expected error for nonexistent agent session")
}
err = repo.UpdateAgentSession(ctx, &models.AgentSession{ID: "nonexistent", TaskID: "task-123"})
err = repo.UpdateTaskSession(ctx, &models.TaskSession{ID: "nonexistent", TaskID: "task-123"})
if err == nil {
t.Error("expected error for updating nonexistent agent session")
}
err = repo.DeleteAgentSession(ctx, "nonexistent")
err = repo.DeleteTaskSession(ctx, "nonexistent")
if err == nil {
t.Error("expected error for deleting nonexistent agent session")
}
}
func TestSQLiteRepository_AgentSessionByTaskID(t *testing.T) {
func TestSQLiteRepository_TaskSessionByTaskID(t *testing.T) {
repo, cleanup := createTestSQLiteRepo(t)
defer cleanup()
ctx := context.Background()
@@ -831,24 +831,24 @@ func TestSQLiteRepository_AgentSessionByTaskID(t *testing.T) {
_ = repo.CreateTask(ctx, task)
// Create multiple sessions for the same task (simulating session history)
session1 := &models.AgentSession{
session1 := &models.TaskSession{
ID: "session-1",
TaskID: "task-123",
AgentProfileID: "profile-1",
State: models.AgentSessionStateCompleted,
State: models.TaskSessionStateCompleted,
}
_ = repo.CreateAgentSession(ctx, session1)
_ = repo.CreateTaskSession(ctx, session1)
session2 := &models.AgentSession{
session2 := &models.TaskSession{
ID: "session-2",
TaskID: "task-123",
AgentProfileID: "profile-2",
State: models.AgentSessionStateRunning,
State: models.TaskSessionStateRunning,
}
_ = repo.CreateAgentSession(ctx, session2)
_ = repo.CreateTaskSession(ctx, session2)
// GetAgentSessionByTaskID should return the most recent session
retrieved, err := repo.GetAgentSessionByTaskID(ctx, "task-123")
// GetTaskSessionByTaskID should return the most recent session
retrieved, err := repo.GetTaskSessionByTaskID(ctx, "task-123")
if err != nil {
t.Fatalf("failed to get agent session by task ID: %v", err)
}
@@ -856,35 +856,35 @@ func TestSQLiteRepository_AgentSessionByTaskID(t *testing.T) {
t.Errorf("expected session-2 (most recent), got %s", retrieved.ID)
}
// GetActiveAgentSessionByTaskID should return the active session
active, err := repo.GetActiveAgentSessionByTaskID(ctx, "task-123")
// GetActiveTaskSessionByTaskID should return the active session
active, err := repo.GetActiveTaskSessionByTaskID(ctx, "task-123")
if err != nil {
t.Fatalf("failed to get active agent session by task ID: %v", err)
}
if active.ID != "session-2" {
t.Errorf("expected session-2 (active), got %s", active.ID)
}
if active.State != models.AgentSessionStateRunning {
if active.State != models.TaskSessionStateRunning {
t.Errorf("expected state 'running', got %s", active.State)
}
// Test when no active session exists
session2.State = models.AgentSessionStateCompleted
_ = repo.UpdateAgentSession(ctx, session2)
session2.State = models.TaskSessionStateCompleted
_ = repo.UpdateTaskSession(ctx, session2)
_, err = repo.GetActiveAgentSessionByTaskID(ctx, "task-123")
_, err = repo.GetActiveTaskSessionByTaskID(ctx, "task-123")
if err == nil {
t.Error("expected error when no active session exists")
}
// Test for nonexistent task
_, err = repo.GetAgentSessionByTaskID(ctx, "nonexistent-task")
_, err = repo.GetTaskSessionByTaskID(ctx, "nonexistent-task")
if err == nil {
t.Error("expected error for nonexistent task")
}
}
func TestSQLiteRepository_ListAgentSessions(t *testing.T) {
func TestSQLiteRepository_ListTaskSessions(t *testing.T) {
repo, cleanup := createTestSQLiteRepo(t)
defer cleanup()
ctx := context.Background()
@@ -900,12 +900,12 @@ func TestSQLiteRepository_ListAgentSessions(t *testing.T) {
_ = repo.CreateTask(ctx, task2)
// Create sessions for different tasks
_ = repo.CreateAgentSession(ctx, &models.AgentSession{ID: "session-1", TaskID: "task-1", AgentProfileID: "profile-1", State: models.AgentSessionStateCompleted})
_ = repo.CreateAgentSession(ctx, &models.AgentSession{ID: "session-2", TaskID: "task-1", AgentProfileID: "profile-1", State: models.AgentSessionStateRunning})
_ = repo.CreateAgentSession(ctx, &models.AgentSession{ID: "session-3", TaskID: "task-2", AgentProfileID: "profile-2", State: models.AgentSessionStateStarting})
_ = repo.CreateTaskSession(ctx, &models.TaskSession{ID: "session-1", TaskID: "task-1", AgentProfileID: "profile-1", State: models.TaskSessionStateCompleted})
_ = repo.CreateTaskSession(ctx, &models.TaskSession{ID: "session-2", TaskID: "task-1", AgentProfileID: "profile-1", State: models.TaskSessionStateRunning})
_ = repo.CreateTaskSession(ctx, &models.TaskSession{ID: "session-3", TaskID: "task-2", AgentProfileID: "profile-2", State: models.TaskSessionStateStarting})
// List sessions for task-1
sessions, err := repo.ListAgentSessions(ctx, "task-1")
sessions, err := repo.ListTaskSessions(ctx, "task-1")
if err != nil {
t.Fatalf("failed to list agent sessions: %v", err)
}
@@ -914,7 +914,7 @@ func TestSQLiteRepository_ListAgentSessions(t *testing.T) {
}
// List all active sessions
activeSessions, err := repo.ListActiveAgentSessions(ctx)
activeSessions, err := repo.ListActiveTaskSessions(ctx)
if err != nil {
t.Fatalf("failed to list active agent sessions: %v", err)
}
@@ -924,13 +924,13 @@ func TestSQLiteRepository_ListAgentSessions(t *testing.T) {
// Verify only active statuses are returned
for _, s := range activeSessions {
if s.State != models.AgentSessionStateStarting && s.State != models.AgentSessionStateRunning && s.State != models.AgentSessionStateWaitingForInput {
if s.State != models.TaskSessionStateStarting && s.State != models.TaskSessionStateRunning && s.State != models.TaskSessionStateWaitingForInput {
t.Errorf("expected active state, got %s", s.State)
}
}
}
func TestSQLiteRepository_UpdateAgentSessionState(t *testing.T) {
func TestSQLiteRepository_UpdateTaskSessionState(t *testing.T) {
repo, cleanup := createTestSQLiteRepo(t)
defer cleanup()
ctx := context.Background()
@@ -944,21 +944,21 @@ func TestSQLiteRepository_UpdateAgentSessionState(t *testing.T) {
_ = repo.CreateTask(ctx, task)
// Create an agent session
session := &models.AgentSession{
session := &models.TaskSession{
ID: "session-123",
TaskID: "task-123",
AgentProfileID: "profile-1",
State: models.AgentSessionStateStarting,
State: models.TaskSessionStateStarting,
}
_ = repo.CreateAgentSession(ctx, session)
_ = repo.CreateTaskSession(ctx, session)
// Update to running status
err := repo.UpdateAgentSessionState(ctx, "session-123", models.AgentSessionStateRunning, "")
err := repo.UpdateTaskSessionState(ctx, "session-123", models.TaskSessionStateRunning, "")
if err != nil {
t.Fatalf("failed to update agent session status: %v", err)
}
retrieved, _ := repo.GetAgentSession(ctx, "session-123")
if retrieved.State != models.AgentSessionStateRunning {
retrieved, _ := repo.GetTaskSession(ctx, "session-123")
if retrieved.State != models.TaskSessionStateRunning {
t.Errorf("expected state 'running', got %s", retrieved.State)
}
if retrieved.CompletedAt != nil {
@@ -966,12 +966,12 @@ func TestSQLiteRepository_UpdateAgentSessionState(t *testing.T) {
}
// Update to completed status (should set CompletedAt)
err = repo.UpdateAgentSessionState(ctx, "session-123", models.AgentSessionStateCompleted, "")
err = repo.UpdateTaskSessionState(ctx, "session-123", models.TaskSessionStateCompleted, "")
if err != nil {
t.Fatalf("failed to update agent session status to completed: %v", err)
}
retrieved, _ = repo.GetAgentSession(ctx, "session-123")
if retrieved.State != models.AgentSessionStateCompleted {
retrieved, _ = repo.GetTaskSession(ctx, "session-123")
if retrieved.State != models.TaskSessionStateCompleted {
t.Errorf("expected state 'completed', got %s", retrieved.State)
}
if retrieved.CompletedAt == nil {
@@ -979,20 +979,20 @@ func TestSQLiteRepository_UpdateAgentSessionState(t *testing.T) {
}
// Test failed status with error message
session2 := &models.AgentSession{
session2 := &models.TaskSession{
ID: "session-456",
TaskID: "task-123",
AgentProfileID: "profile-1",
State: models.AgentSessionStateRunning,
State: models.TaskSessionStateRunning,
}
_ = repo.CreateAgentSession(ctx, session2)
_ = repo.CreateTaskSession(ctx, session2)
err = repo.UpdateAgentSessionState(ctx, "session-456", models.AgentSessionStateFailed, "connection timeout")
err = repo.UpdateTaskSessionState(ctx, "session-456", models.TaskSessionStateFailed, "connection timeout")
if err != nil {
t.Fatalf("failed to update agent session status to failed: %v", err)
}
retrieved, _ = repo.GetAgentSession(ctx, "session-456")
if retrieved.State != models.AgentSessionStateFailed {
retrieved, _ = repo.GetTaskSession(ctx, "session-456")
if retrieved.State != models.TaskSessionStateFailed {
t.Errorf("expected state 'failed', got %s", retrieved.State)
}
if retrieved.ErrorMessage != "connection timeout" {
@@ -1003,20 +1003,20 @@ func TestSQLiteRepository_UpdateAgentSessionState(t *testing.T) {
}
// Test stopped status
session3 := &models.AgentSession{
session3 := &models.TaskSession{
ID: "session-789",
TaskID: "task-123",
AgentProfileID: "profile-1",
State: models.AgentSessionStateRunning,
State: models.TaskSessionStateRunning,
}
_ = repo.CreateAgentSession(ctx, session3)
_ = repo.CreateTaskSession(ctx, session3)
err = repo.UpdateAgentSessionState(ctx, "session-789", models.AgentSessionStateCancelled, "")
err = repo.UpdateTaskSessionState(ctx, "session-789", models.TaskSessionStateCancelled, "")
if err != nil {
t.Fatalf("failed to update agent session status to stopped: %v", err)
}
retrieved, _ = repo.GetAgentSession(ctx, "session-789")
if retrieved.State != models.AgentSessionStateCancelled {
retrieved, _ = repo.GetTaskSession(ctx, "session-789")
if retrieved.State != models.TaskSessionStateCancelled {
t.Errorf("expected state 'cancelled', got %s", retrieved.State)
}
if retrieved.CompletedAt == nil {
@@ -1024,7 +1024,7 @@ func TestSQLiteRepository_UpdateAgentSessionState(t *testing.T) {
}
// Test nonexistent session
err = repo.UpdateAgentSessionState(ctx, "nonexistent", models.AgentSessionStateRunning, "")
err = repo.UpdateTaskSessionState(ctx, "nonexistent", models.TaskSessionStateRunning, "")
if err == nil {
t.Error("expected error for updating nonexistent session status")
}
apps/backend/internal/task/service/service.go
+14 -14
View File
@@ -29,7 +29,7 @@ type TaskExecutionStopper interface {
StopTask(ctx context.Context, taskID, reason string, force bool) error
}
var ErrActiveAgentSessions = errors.New("active agent sessions exist")
var ErrActiveTaskSessions = errors.New("active agent sessions exist")
// Service provides task business logic
type Service struct {
@@ -208,7 +208,7 @@ type UpdateEnvironmentRequest struct {
}
type ListMessagesRequest struct {
AgentSessionID string
TaskSessionID string
Limit int
Before string
After string
@@ -839,13 +839,13 @@ func (s *Service) DeleteRepository(ctx context.Context, id string) error {
if err != nil {
return err
}
active, err := s.repo.HasActiveAgentSessionsByRepository(ctx, id)
active, err := s.repo.HasActiveTaskSessionsByRepository(ctx, id)
if err != nil {
s.logger.Error("failed to check active agent sessions for repository", zap.String("repository_id", id), zap.Error(err))
return err
}
if active {
return ErrActiveAgentSessions
return ErrActiveTaskSessions
}
if err := s.repo.DeleteRepository(ctx, id); err != nil {
s.logger.Error("failed to delete repository", zap.String("repository_id", id), zap.Error(err))
@@ -985,13 +985,13 @@ func (s *Service) DeleteExecutor(ctx context.Context, id string) error {
if executor.IsSystem {
return fmt.Errorf("system executors cannot be deleted")
}
active, err := s.repo.HasActiveAgentSessionsByExecutor(ctx, id)
active, err := s.repo.HasActiveTaskSessionsByExecutor(ctx, id)
if err != nil {
s.logger.Error("failed to check active agent sessions for executor", zap.String("executor_id", id), zap.Error(err))
return err
}
if active {
return ErrActiveAgentSessions
return ErrActiveTaskSessions
}
if err := s.repo.DeleteExecutor(ctx, id); err != nil {
return err
@@ -1072,13 +1072,13 @@ func (s *Service) DeleteEnvironment(ctx context.Context, id string) error {
if environment.IsSystem {
return fmt.Errorf("system environments cannot be deleted")
}
active, err := s.repo.HasActiveAgentSessionsByEnvironment(ctx, id)
active, err := s.repo.HasActiveTaskSessionsByEnvironment(ctx, id)
if err != nil {
s.logger.Error("failed to check active agent sessions for environment", zap.String("environment_id", id), zap.Error(err))
return err
}
if active {
return ErrActiveAgentSessions
return ErrActiveTaskSessions
}
if err := s.repo.DeleteEnvironment(ctx, id); err != nil {
return err
@@ -1276,7 +1276,7 @@ func (s *Service) publishEnvironmentEvent(ctx context.Context, eventType string,
// CreateMessageRequest contains the data for creating a new message
type CreateMessageRequest struct {
AgentSessionID string `json:"agent_session_id"`
TaskSessionID string `json:"agent_session_id"`
TaskID string `json:"task_id,omitempty"`
Content string `json:"content"`
AuthorType string `json:"author_type,omitempty"` // "user" or "agent", defaults to "user"
@@ -1288,7 +1288,7 @@ type CreateMessageRequest struct {
// CreateMessage creates a new message on an agent session
func (s *Service) CreateMessage(ctx context.Context, req *CreateMessageRequest) (*models.Message, error) {
session, err := s.repo.GetAgentSession(ctx, req.AgentSessionID)
session, err := s.repo.GetTaskSession(ctx, req.TaskSessionID)
if err != nil {
return nil, err
}
@@ -1310,7 +1310,7 @@ func (s *Service) CreateMessage(ctx context.Context, req *CreateMessageRequest)
message := &models.Message{
ID: uuid.New().String(),
AgentSessionID: req.AgentSessionID,
TaskSessionID: req.TaskSessionID,
TaskID: taskID,
AuthorType: authorType,
AuthorID: req.AuthorID,
@@ -1331,7 +1331,7 @@ func (s *Service) CreateMessage(ctx context.Context, req *CreateMessageRequest)
s.logger.Info("message created",
zap.String("message_id", message.ID),
zap.String("agent_session_id", message.AgentSessionID),
zap.String("agent_session_id", message.TaskSessionID),
zap.String("author_type", string(message.AuthorType)))
return message, nil
@@ -1356,7 +1356,7 @@ func (s *Service) ListMessagesPaginated(ctx context.Context, req ListMessagesReq
if limit > MaxMessagesPageSize {
limit = MaxMessagesPageSize
}
return s.repo.ListMessagesPaginated(ctx, req.AgentSessionID, repository.ListMessagesOptions{
return s.repo.ListMessagesPaginated(ctx, req.TaskSessionID, repository.ListMessagesOptions{
Limit: limit,
Before: req.Before,
After: req.After,
@@ -1427,7 +1427,7 @@ func (s *Service) publishMessageEvent(ctx context.Context, eventType string, mes
data := map[string]interface{}{
"message_id": message.ID,
"agent_session_id": message.AgentSessionID,
"agent_session_id": message.TaskSessionID,
"task_id": message.TaskID,
"author_type": string(message.AuthorType),
"author_id": message.AuthorID,
apps/backend/internal/task/service/service_test.go
+10 -10
View File
@@ -472,13 +472,13 @@ func setupTestTask(t *testing.T, repo repository.Repository) {
func setupTestSession(t *testing.T, repo repository.Repository) string {
t.Helper()
ctx := context.Background()
session := &models.AgentSession{
session := &models.TaskSession{
ID: "session-123",
TaskID: "task-123",
AgentProfileID: "profile-123",
State: models.AgentSessionStateStarting,
State: models.TaskSessionStateStarting,
}
_ = repo.CreateAgentSession(ctx, session)
_ = repo.CreateTaskSession(ctx, session)
return session.ID
}
@@ -492,7 +492,7 @@ func TestService_CreateMessage(t *testing.T) {
eventBus.ClearEvents()
req := &CreateMessageRequest{
AgentSessionID: sessionID,
TaskSessionID: sessionID,
Content: "This is a test comment",
AuthorType: "user",
AuthorID: "user-123",
@@ -531,7 +531,7 @@ func TestService_CreateAgentMessage(t *testing.T) {
sessionID := setupTestSession(t, repo)
req := &CreateMessageRequest{
AgentSessionID: sessionID,
TaskSessionID: sessionID,
Content: "What should I do next?",
AuthorType: "agent",
AuthorID: "agent-123",
@@ -556,7 +556,7 @@ func TestService_CreateMessageSessionNotFound(t *testing.T) {
ctx := context.Background()
req := &CreateMessageRequest{
AgentSessionID: "nonexistent",
TaskSessionID: "nonexistent",
Content: "Test comment",
}
@@ -573,7 +573,7 @@ func TestService_GetMessage(t *testing.T) {
setupTestTask(t, repo)
sessionID := setupTestSession(t, repo)
comment := &models.Message{ID: "comment-123", AgentSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorUser, Content: "Test"}
comment := &models.Message{ID: "comment-123", TaskSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorUser, Content: "Test"}
_ = repo.CreateMessage(ctx, comment)
retrieved, err := svc.GetMessage(ctx, "comment-123")
@@ -592,8 +592,8 @@ func TestService_ListMessages(t *testing.T) {
setupTestTask(t, repo)
sessionID := setupTestSession(t, repo)
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-1", AgentSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorUser, Content: "Comment 1"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-2", AgentSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorAgent, Content: "Comment 2"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-1", TaskSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorUser, Content: "Comment 1"})
_ = repo.CreateMessage(ctx, &models.Message{ID: "comment-2", TaskSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorAgent, Content: "Comment 2"})
comments, err := svc.ListMessages(ctx, sessionID)
if err != nil {
@@ -611,7 +611,7 @@ func TestService_DeleteMessage(t *testing.T) {
setupTestTask(t, repo)
sessionID := setupTestSession(t, repo)
comment := &models.Message{ID: "comment-123", AgentSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorUser, Content: "Test"}
comment := &models.Message{ID: "comment-123", TaskSessionID: sessionID, TaskID: "task-123", AuthorType: models.MessageAuthorUser, Content: "Test"}
_ = repo.CreateMessage(ctx, comment)
err := svc.DeleteMessage(ctx, "comment-123")
apps/backend/pkg/api/v1/task.go
+11 -11
View File
@@ -18,17 +18,17 @@ const (
TaskStateCancelled TaskState = "CANCELLED"
)
// AgentSessionState represents the state of an agent session.
type AgentSessionState string
// TaskSessionState represents the state of an agent session.
type TaskSessionState string
const (
AgentSessionStateCreated AgentSessionState = "CREATED"
AgentSessionStateStarting AgentSessionState = "STARTING"
AgentSessionStateRunning AgentSessionState = "RUNNING"
AgentSessionStateWaitingForInput AgentSessionState = "WAITING_FOR_INPUT"
AgentSessionStateCompleted AgentSessionState = "COMPLETED"
AgentSessionStateFailed AgentSessionState = "FAILED"
AgentSessionStateCancelled AgentSessionState = "CANCELLED"
TaskSessionStateCreated TaskSessionState = "CREATED"
TaskSessionStateStarting TaskSessionState = "STARTING"
TaskSessionStateRunning TaskSessionState = "RUNNING"
TaskSessionStateWaitingForInput TaskSessionState = "WAITING_FOR_INPUT"
TaskSessionStateCompleted TaskSessionState = "COMPLETED"
TaskSessionStateFailed TaskSessionState = "FAILED"
TaskSessionStateCancelled TaskSessionState = "CANCELLED"
)
// MessageType represents a normalized session message type.
@@ -105,7 +105,7 @@ type TaskEvent struct {
// Message represents a message in an agent session (user or agent)
type Message struct {
ID string `json:"id"`
AgentSessionID string `json:"agent_session_id"`
TaskSessionID string `json:"agent_session_id"`
TaskID string `json:"task_id,omitempty"`
AuthorType string `json:"author_type"` // "user" or "agent"
Type string `json:"type,omitempty"`
@@ -118,7 +118,7 @@ type Message struct {
// CreateMessageRequest for adding a message to an agent session
type CreateMessageRequest struct {
AgentSessionID string `json:"agent_session_id" binding:"required"`
TaskSessionID string `json:"agent_session_id" binding:"required"`
Content string `json:"content" binding:"required"`
AuthorType string `json:"author_type,omitempty"` // Defaults to "user" if not specified
Type string `json:"type,omitempty"`
apps/backend/pkg/websocket/actions.go
+2 -2
View File
@@ -72,7 +72,7 @@ const (
ActionAgentStop = "agent.stop"
ActionAgentPrompt = "agent.prompt"
ActionAgentCancel = "agent.cancel"
ActionAgentSession = "agent.session"
ActionTaskSession = "task.session"
ActionAgentTypes = "agent.types"
// Orchestrator actions
@@ -120,7 +120,7 @@ const (
ActionColumnDeleted = "column.deleted"
ActionMessageAdded = "message.added"
ActionMessageUpdated = "message.updated"
ActionAgentSessionStateChanged = "agent_session.state_changed"
ActionTaskSessionStateChanged = "task_session.state_changed"
ActionInputRequested = "input.requested"
ActionRepositoryCreated = "repository.created"
ActionRepositoryUpdated = "repository.updated"
apps/web/app/task/[id]/page-client.tsx
+10 -10
View File
@@ -44,16 +44,16 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) {
isAgentLoading,
worktreePath,
worktreeBranch,
agentSessionId,
agentSessionState,
taskSessionId,
taskSessionState,
handleStartAgent,
handleStopAgent,
} = useTaskAgent(task);
const isAgentWorking =
agentSessionState !== null
? agentSessionState === 'STARTING' || agentSessionState === 'RUNNING'
taskSessionState !== null
? taskSessionState === 'STARTING' || taskSessionState === 'RUNNING'
: isAgentRunning && (task?.state === 'IN_PROGRESS' || task?.state === 'SCHEDULING');
const { isLoading: isLoadingMessages } = useTaskMessages(task?.id ?? null, agentSessionId);
const { isLoading: isLoadingMessages } = useTaskMessages(task?.id ?? null, taskSessionId);
useEffect(() => {
queueMicrotask(() => setIsMounted(true));
@@ -71,7 +71,7 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) {
const client = getWebSocketClient();
if (!client) return;
if (!agentSessionId) {
if (!taskSessionId) {
console.error('No active agent session. Start an agent before sending a message.');
return;
}
@@ -79,13 +79,13 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) {
try {
await client.request(
'message.add',
{ task_id: task.id, agent_session_id: agentSessionId, content },
{ task_id: task.id, task_session_id: taskSessionId, content },
10000
);
} catch (error) {
console.error('Failed to send message:', error);
}
}, [agentSessionId, task]);
}, [taskSessionId, task]);
if (!isMounted) {
return <div className="h-screen w-full bg-background" />;
@@ -100,9 +100,9 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) {
entries={{
ws_status: connectionStatus,
task_id: task?.id ?? null,
agent_session_id: agentSessionId ?? null,
task_session_id: taskSessionId ?? null,
task_state: task?.state ?? null,
agent_session_state: agentSessionState ?? null,
agent_session_state: taskSessionState ?? null,
is_agent_working: isAgentWorking,
}}
/>
apps/web/components/task/task-chat-panel.tsx
+1 -1
View File
@@ -82,7 +82,7 @@ export function TaskChatPanel({
? {
id: 'task-description',
task_id: taskId ?? '',
agent_session_id: messagesState?.sessionId ?? '',
task_session_id: messagesState?.sessionId ?? '',
author_type: 'user',
content: taskDescription,
type: 'message',
apps/web/hooks/use-lazy-load-messages.ts
+2 -2
View File
@@ -1,5 +1,5 @@
import { useCallback } from 'react';
import { listAgentSessionMessages } from '@/lib/http';
import { listTaskSessionMessages } from '@/lib/http';
import { useAppStore } from '@/components/state-provider';
export function useLazyLoadMessages(sessionId: string | null) {
@@ -14,7 +14,7 @@ export function useLazyLoadMessages(sessionId: string | null) {
setMessagesMetadata({ isLoading: true });
try {
const response = await listAgentSessionMessages(sessionId, {
const response = await listTaskSessionMessages(sessionId, {
limit: 20,
before: oldestCursor,
sort: 'desc',
apps/web/hooks/use-task-agent.ts
+23 -23
View File
@@ -1,13 +1,13 @@
import { useCallback, useEffect, useState } from 'react';
import { getWebSocketClient } from '@/lib/ws/connection';
import { useAppStore } from '@/components/state-provider';
import type { AgentSessionState, Task } from '@/lib/types/http';
import type { TaskSessionState, Task } from '@/lib/types/http';
interface UseTaskAgentReturn {
isAgentRunning: boolean;
isAgentLoading: boolean;
agentSessionId: string | null;
agentSessionState: AgentSessionState | null;
taskSessionId: string | null;
taskSessionState: TaskSessionState | null;
worktreePath: string | null;
worktreeBranch: string | null;
handleStartAgent: (agentProfileId: string) => Promise<void>;
@@ -17,27 +17,27 @@ interface UseTaskAgentReturn {
export function useTaskAgent(task: Task | null): UseTaskAgentReturn {
const [isAgentRunning, setIsAgentRunning] = useState(false);
const [isAgentLoading, setIsAgentLoading] = useState(false);
const [agentSessionId, setAgentSessionId] = useState<string | null>(null);
const [agentSessionState, setAgentSessionState] = useState<AgentSessionState | null>(null);
const [taskSessionId, setAgentSessionId] = useState<string | null>(null);
const [taskSessionState, setTaskSessionState] = useState<TaskSessionState | null>(null);
const [worktreePath, setWorktreePath] = useState<string | null>(task?.worktree_path ?? null);
const [worktreeBranch, setWorktreeBranch] = useState<string | null>(task?.worktree_branch ?? null);
const connectionStatus = useAppStore((state) => state.connection.status);
const wsAgentSessionState = useAppStore((state) =>
task?.id ? state.agentSessionStatesByTaskId[task.id] : undefined
const wsTaskSessionState = useAppStore((state) =>
task?.id ? state.taskSessionStatesByTaskId[task.id] : undefined
);
useEffect(() => {
setIsAgentRunning(false);
setAgentSessionId(null);
setAgentSessionState(null);
setTaskSessionState(null);
setWorktreePath(task?.worktree_path ?? null);
setWorktreeBranch(task?.worktree_branch ?? null);
}, [task?.id, task?.worktree_path, task?.worktree_branch]);
useEffect(() => {
if (!wsAgentSessionState) return;
setAgentSessionState(wsAgentSessionState as AgentSessionState);
}, [wsAgentSessionState]);
if (!wsTaskSessionState) return;
setTaskSessionState(wsTaskSessionState as TaskSessionState);
}, [wsTaskSessionState]);
// Fetch task execution status from orchestrator on mount
useEffect(() => {
@@ -53,22 +53,22 @@ export function useTaskAgent(task: Task | null): UseTaskAgentReturn {
has_execution: boolean;
task_id: string;
state?: string;
agent_session_id?: string;
task_session_id?: string;
}>('task.execution', { task_id: task.id });
console.log('[useTaskAgent] Task execution check:', response);
if (response.has_execution) {
setIsAgentRunning(true);
if (response.state) {
setAgentSessionState(response.state as AgentSessionState);
setTaskSessionState(response.state as TaskSessionState);
}
if (response.agent_session_id) {
setAgentSessionId(response.agent_session_id);
if (response.task_session_id) {
setAgentSessionId(response.task_session_id);
}
} else {
setIsAgentRunning(false);
setAgentSessionId(null);
setAgentSessionState(null);
setTaskSessionState(null);
}
} catch (err) {
console.error('[useTaskAgent] Failed to check task execution:', err);
@@ -98,7 +98,7 @@ export function useTaskAgent(task: Task | null): UseTaskAgentReturn {
success: boolean;
task_id: string;
agent_instance_id: string;
agent_session_id?: string;
task_session_id?: string;
state: string;
worktree_path?: string;
worktree_branch?: string;
@@ -113,9 +113,9 @@ export function useTaskAgent(task: Task | null): UseTaskAgentReturn {
}, 15000);
console.log('[useTaskAgent] orchestrator.start response', response);
setIsAgentRunning(true);
setAgentSessionState(response.state as AgentSessionState);
if (response?.agent_session_id) {
setAgentSessionId(response.agent_session_id);
setTaskSessionState(response.state as TaskSessionState);
if (response?.task_session_id) {
setAgentSessionId(response.task_session_id);
}
// Update worktree info from response
@@ -141,7 +141,7 @@ export function useTaskAgent(task: Task | null): UseTaskAgentReturn {
await client.request('orchestrator.stop', { task_id: task.id }, 15000);
setIsAgentRunning(false);
setAgentSessionId(null);
setAgentSessionState(null);
setTaskSessionState(null);
} catch (error) {
console.error('Failed to stop agent:', error);
} finally {
@@ -152,8 +152,8 @@ export function useTaskAgent(task: Task | null): UseTaskAgentReturn {
return {
isAgentRunning,
isAgentLoading,
agentSessionId,
agentSessionState,
taskSessionId,
taskSessionState,
worktreePath,
worktreeBranch,
handleStartAgent,
apps/web/hooks/use-task-messages.ts
+28 -28
View File
@@ -1,7 +1,7 @@
import { useEffect, useRef, useState } from 'react';
import { getWebSocketClient } from '@/lib/ws/connection';
import { useAppStore, useAppStoreApi } from '@/components/state-provider';
import type { AgentSessionState, Message } from '@/lib/types/http';
import type { TaskSessionState, Message } from '@/lib/types/http';
interface UseTaskMessagesReturn {
isLoading: boolean;
@@ -9,12 +9,12 @@ interface UseTaskMessagesReturn {
export function useTaskMessages(
taskId: string | null,
agentSessionId: string | null
taskSessionId: string | null
): UseTaskMessagesReturn {
const store = useAppStoreApi();
const messagesState = useAppStore((state) => state.messages);
const agentSessionState = useAppStore((state) =>
taskId ? (state.agentSessionStatesByTaskId[taskId] ?? null) : null
const taskSessionState = useAppStore((state) =>
taskId ? (state.taskSessionStatesByTaskId[taskId] ?? null) : null
);
const connectionStatus = useAppStore((state) => state.connection.status);
const [isLoading, setIsLoading] = useState(false);
@@ -30,28 +30,28 @@ export function useTaskMessages(
}, [store, taskId]);
useEffect(() => {
store.getState().setMessagesSessionId(agentSessionId);
if (!agentSessionId) {
console.log('[useTaskMessages] no agent_session_id yet, clearing messages');
store.getState().setMessagesSessionId(taskSessionId);
if (!taskSessionId) {
console.log('[useTaskMessages] no task_session_id yet, clearing messages');
store.getState().setMessages(null, []);
initialFetchStartRef.current = null;
lastFetchedSessionIdRef.current = null;
setIsWaitingForInitialMessages(false);
}
}, [agentSessionId, store]);
}, [taskSessionId, store]);
useEffect(() => {
if (!agentSessionId) return;
if (!taskSessionId) return;
if (messagesState.items.length > 0) return;
if (initialFetchStartRef.current === null) {
initialFetchStartRef.current = Date.now();
setIsWaitingForInitialMessages(true);
}
}, [agentSessionId, messagesState.items.length]);
}, [taskSessionId, messagesState.items.length]);
// Fetch messages on mount and when session changes
useEffect(() => {
if (!agentSessionId) return;
if (!taskSessionId) return;
if (connectionStatus !== 'connected') {
console.warn('[useTaskMessages] WebSocket not connected yet, waiting to fetch messages');
return;
@@ -59,9 +59,9 @@ export function useTaskMessages(
// Set sessionId immediately so that incoming WebSocket notifications are processed
// before the API call completes (fixes race condition on first agent start)
store.getState().setMessagesSessionId(agentSessionId);
store.getState().setMessagesSessionId(taskSessionId);
if (lastFetchedSessionIdRef.current === agentSessionId && messagesState.items.length > 0) {
if (lastFetchedSessionIdRef.current === taskSessionId && messagesState.items.length > 0) {
return;
}
@@ -80,29 +80,29 @@ export function useTaskMessages(
}
try {
console.log('[useTaskMessages] requesting message.list', { agentSessionId });
console.log('[useTaskMessages] requesting message.list', { taskSessionId });
const response = await client.request<{ messages: Message[] }>(
'message.list',
{ agent_session_id: agentSessionId },
{ task_session_id: taskSessionId },
10000
);
console.log('[useTaskMessages] message.list response:', JSON.stringify(response, null, 2));
store.getState().setMessages(agentSessionId, response.messages ?? []);
lastFetchedSessionIdRef.current = agentSessionId;
store.getState().setMessages(taskSessionId, response.messages ?? []);
lastFetchedSessionIdRef.current = taskSessionId;
if ((response.messages ?? []).length > 0) {
setIsWaitingForInitialMessages(false);
}
} catch (error) {
console.error('Failed to fetch messages:', error);
store.getState().setMessages(agentSessionId, []);
lastFetchedSessionIdRef.current = agentSessionId;
store.getState().setMessages(taskSessionId, []);
lastFetchedSessionIdRef.current = taskSessionId;
} finally {
setIsLoading(false);
}
};
fetchMessages();
}, [agentSessionId, connectionStatus, messagesState.items.length, store]);
}, [taskSessionId, connectionStatus, messagesState.items.length, store]);
// Subscribe to task for real-time updates
useEffect(() => {
@@ -126,15 +126,15 @@ export function useTaskMessages(
}, [taskId]);
useEffect(() => {
if (!agentSessionId || !agentSessionState || !taskId) return;
if (!taskSessionId || !taskSessionState || !taskId) return;
if (hasAgentMessage) return;
const terminalStates = new Set<AgentSessionState>(['WAITING_FOR_INPUT', 'COMPLETED', 'FAILED']);
if (!terminalStates.has(agentSessionState)) {
const terminalStates = new Set<TaskSessionState>(['WAITING_FOR_INPUT', 'COMPLETED', 'FAILED']);
if (!terminalStates.has(taskSessionState)) {
return;
}
const key = `${agentSessionId}:${agentSessionState}`;
const key = `${taskSessionId}:${taskSessionState}`;
if (lastFetchStateKeyRef.current === key) {
return;
}
@@ -150,13 +150,13 @@ export function useTaskMessages(
setIsLoading(true);
store.getState().setMessagesLoading(true);
try {
console.log('[useTaskMessages] requesting message.list after state change', { agentSessionId, agentSessionState });
console.log('[useTaskMessages] requesting message.list after state change', { taskSessionId, taskSessionState });
const response = await client.request<{ messages: Message[] }>(
'message.list',
{ agent_session_id: agentSessionId },
{ task_session_id: taskSessionId },
10000
);
store.getState().setMessages(agentSessionId, response.messages ?? []);
store.getState().setMessages(taskSessionId, response.messages ?? []);
if ((response.messages ?? []).length > 0) {
setIsWaitingForInitialMessages(false);
}
@@ -168,7 +168,7 @@ export function useTaskMessages(
};
fetchMessages();
}, [agentSessionId, agentSessionState, hasAgentMessage, store, taskId]);
}, [taskSessionId, taskSessionState, hasAgentMessage, store, taskId]);
return {
isLoading: isLoading || isWaitingForInitialMessages,
apps/web/lib/http/index.ts
+3 -3
View File
@@ -159,8 +159,8 @@ export async function fetchTask(taskId: string, options?: ApiRequestOptions) {
return fetchJson<Task>(`/api/v1/tasks/${taskId}`, options);
}
export async function listAgentSessionMessages(
agentSessionId: string,
export async function listTaskSessionMessages(
taskSessionId: string,
params?: { limit?: number; before?: string; after?: string; sort?: 'asc' | 'desc' },
options?: ApiRequestOptions
) {
@@ -170,7 +170,7 @@ export async function listAgentSessionMessages(
if (params?.after) query.set('after', params.after);
if (params?.sort) query.set('sort', params.sort);
const suffix = query.toString();
const url = `/api/v1/agent-sessions/${agentSessionId}/messages${suffix ? `?${suffix}` : ''}`;
const url = `/api/v1/task-sessions/${taskSessionId}/messages${suffix ? `?${suffix}` : ''}`;
return fetchJson<ListMessagesResponse>(url, options);
}
apps/web/lib/ssr/mapper.ts
+1 -1
View File
@@ -44,7 +44,7 @@ export function taskToState(
},
messages: messages
? {
sessionId: messages.items[0]?.agent_session_id ?? null,
sessionId: messages.items[0]?.task_session_id ?? null,
items: messages.items,
isLoading: false,
hasMore: messages.hasMore ?? false,
apps/web/lib/state/store.ts
+17 -17
View File
@@ -2,7 +2,7 @@ import type { ReactNode } from 'react';
import type {
Agent,
AgentDiscovery,
AgentSessionState,
TaskSessionState,
Branch,
Environment,
Executor,
@@ -172,7 +172,7 @@ export type AppState = {
gitStatus: GitStatusState;
connection: ConnectionState;
messages: MessagesState;
agentSessionStatesByTaskId: Record<string, AgentSessionState>;
taskSessionStatesByTaskId: Record<string, TaskSessionState>;
hydrate: (state: Partial<AppState>) => void;
setActiveWorkspace: (workspaceId: string | null) => void;
setWorkspaces: (workspaces: WorkspaceState['items']) => void;
@@ -202,7 +202,7 @@ export type AppState = {
prependMessages: (messages: Message[], meta?: { hasMore?: boolean; oldestCursor?: string | null }) => void;
setMessagesMetadata: (meta: { hasMore?: boolean; isLoading?: boolean; oldestCursor?: string | null }) => void;
setMessagesLoading: (loading: boolean) => void;
setAgentSessionState: (taskId: string, state: AgentSessionState) => void;
setTaskSessionState: (taskId: string, state: TaskSessionState) => void;
setGitStatus: (taskId: string, gitStatus: Omit<GitStatusState, 'taskId'>) => void;
clearGitStatus: () => void;
bumpAgentProfilesVersion: () => void;
@@ -243,7 +243,7 @@ const defaultState: AppState = {
},
connection: { status: 'disconnected', error: null },
messages: { sessionId: null, items: [], isLoading: false, hasMore: false, oldestCursor: null },
agentSessionStatesByTaskId: {},
taskSessionStatesByTaskId: {},
hydrate: () => undefined,
setActiveWorkspace: () => undefined,
setWorkspaces: () => undefined,
@@ -269,7 +269,7 @@ const defaultState: AppState = {
prependMessages: () => undefined,
setMessagesMetadata: () => undefined,
setMessagesLoading: () => undefined,
setAgentSessionState: () => undefined,
setTaskSessionState: () => undefined,
setGitStatus: () => undefined,
clearGitStatus: () => undefined,
bumpAgentProfilesVersion: () => undefined,
@@ -304,7 +304,7 @@ function mergeInitialState(
| 'prependMessages'
| 'setMessagesMetadata'
| 'setMessagesLoading'
| 'setAgentSessionState'
| 'setTaskSessionState'
| 'setGitStatus'
| 'clearGitStatus'
| 'bumpAgentProfilesVersion'
@@ -330,9 +330,9 @@ function mergeInitialState(
gitStatus: { ...defaultState.gitStatus, ...initialState.gitStatus },
connection: { ...defaultState.connection, ...initialState.connection },
messages: { ...defaultState.messages, ...initialState.messages },
agentSessionStatesByTaskId: {
...defaultState.agentSessionStatesByTaskId,
...initialState.agentSessionStatesByTaskId,
taskSessionStatesByTaskId: {
...defaultState.taskSessionStatesByTaskId,
...initialState.taskSessionStatesByTaskId,
},
};
}
@@ -363,8 +363,8 @@ export function createAppStore(initialState?: Partial<AppState>) {
if (state.diffs) Object.assign(draft.diffs, state.diffs);
if (state.connection) Object.assign(draft.connection, state.connection);
if (state.messages) Object.assign(draft.messages, state.messages);
if (state.agentSessionStatesByTaskId) {
Object.assign(draft.agentSessionStatesByTaskId, state.agentSessionStatesByTaskId);
if (state.taskSessionStatesByTaskId) {
Object.assign(draft.taskSessionStatesByTaskId, state.taskSessionStatesByTaskId);
}
}),
setActiveWorkspace: (workspaceId) =>
@@ -487,10 +487,10 @@ export function createAppStore(initialState?: Partial<AppState>) {
draft.messages.items = [];
}
if (!draft.messages.sessionId) {
draft.messages.sessionId = message.agent_session_id;
draft.messages.sessionId = message.task_session_id;
}
// Only add if this message is for the current session
if (draft.messages.sessionId === message.agent_session_id) {
if (draft.messages.sessionId === message.task_session_id) {
// Check if message already exists (avoid duplicates)
const exists = draft.messages.items.some((item) => item.id === message.id);
if (!exists) {
@@ -504,10 +504,10 @@ export function createAppStore(initialState?: Partial<AppState>) {
updateMessage: (message) =>
set((draft) => {
if (!draft.messages.sessionId) {
draft.messages.sessionId = message.agent_session_id;
draft.messages.sessionId = message.task_session_id;
}
// Only update if this message is for the current session
if (draft.messages.sessionId === message.agent_session_id && draft.messages.items) {
if (draft.messages.sessionId === message.task_session_id && draft.messages.items) {
const index = draft.messages.items.findIndex((item) => item.id === message.id);
if (index !== -1) {
draft.messages.items[index] = message;
@@ -552,9 +552,9 @@ export function createAppStore(initialState?: Partial<AppState>) {
set((draft) => {
draft.messages.isLoading = loading;
}),
setAgentSessionState: (taskId, state) =>
setTaskSessionState: (taskId, state) =>
set((draft) => {
draft.agentSessionStatesByTaskId[taskId] = state;
draft.taskSessionStatesByTaskId[taskId] = state;
}),
setGitStatus: (taskId, gitStatus) =>
set((draft) => {
apps/web/lib/types/backend.ts
+5 -5
View File
@@ -20,7 +20,7 @@ export type BackendMessageType =
| 'column.deleted'
| 'message.added'
| 'message.updated'
| 'agent_session.state_changed'
| 'task_session.state_changed'
| 'executor.created'
| 'executor.updated'
| 'executor.deleted'
@@ -130,7 +130,7 @@ export type ColumnPayload = {
export type MessageAddedPayload = {
task_id: string;
message_id: string;
agent_session_id: string;
task_session_id: string;
author_type: 'user' | 'agent';
author_id?: string;
content: string;
@@ -140,9 +140,9 @@ export type MessageAddedPayload = {
created_at: string;
};
export type AgentSessionStateChangedPayload = {
export type TaskSessionStateChangedPayload = {
task_id: string;
agent_session_id: string;
task_session_id: string;
old_state?: string;
new_state: string;
};
@@ -245,7 +245,7 @@ export type BackendMessageMap = {
'column.deleted': BackendMessage<'column.deleted', ColumnPayload>;
'message.added': BackendMessage<'message.added', MessageAddedPayload>;
'message.updated': BackendMessage<'message.updated', MessageAddedPayload>;
'agent_session.state_changed': BackendMessage<'agent_session.state_changed', AgentSessionStateChangedPayload>;
'task_session.state_changed': BackendMessage<'task_session.state_changed', TaskSessionStateChangedPayload>;
'executor.created': BackendMessage<'executor.created', ExecutorPayload>;
'executor.updated': BackendMessage<'executor.updated', ExecutorPayload>;
'executor.deleted': BackendMessage<'executor.deleted', ExecutorPayload>;
apps/web/lib/types/http.ts
+2 -2
View File
@@ -10,7 +10,7 @@ export type TaskState =
| 'FAILED'
| 'CANCELLED';
export type AgentSessionState =
export type TaskSessionState =
| 'CREATED'
| 'STARTING'
| 'RUNNING'
@@ -240,7 +240,7 @@ export type MessageType =
export type Message = {
id: string;
agent_session_id: string;
task_session_id: string;
task_id: string;
author_type: MessageAuthorType;
author_id?: string;
apps/web/lib/ws/handlers/agent-session.ts
+4 -4
View File
@@ -1,16 +1,16 @@
import type { StoreApi } from 'zustand';
import type { AppState } from '@/lib/state/store';
import type { WsHandlers } from '@/lib/ws/handlers/types';
import type { AgentSessionState } from '@/lib/types/http';
import type { TaskSessionState } from '@/lib/types/http';
export function registerAgentSessionHandlers(store: StoreApi<AppState>): WsHandlers {
export function registerTaskSessionHandlers(store: StoreApi<AppState>): WsHandlers {
return {
'agent_session.state_changed': (message) => {
'task_session.state_changed': (message) => {
const payload = message.payload;
if (!payload?.task_id || !payload?.new_state) {
return;
}
store.getState().setAgentSessionState(payload.task_id, payload.new_state as AgentSessionState);
store.getState().setTaskSessionState(payload.task_id, payload.new_state as TaskSessionState);
},
};
}
apps/web/lib/ws/handlers/messages.ts
+2 -2
View File
@@ -14,7 +14,7 @@ export function registerMessagesHandlers(store: StoreApi<AppState>): WsHandlers
});
state.addMessage({
id: payload.message_id,
agent_session_id: payload.agent_session_id,
task_session_id: payload.task_session_id,
task_id: payload.task_id,
author_type: payload.author_type,
author_id: payload.author_id,
@@ -32,7 +32,7 @@ export function registerMessagesHandlers(store: StoreApi<AppState>): WsHandlers
const state = store.getState();
state.updateMessage({
id: payload.message_id,
agent_session_id: payload.agent_session_id,
task_session_id: payload.task_session_id,
task_id: payload.task_id,
author_type: payload.author_type,
author_id: payload.author_id,
apps/web/lib/ws/router.ts
+2 -2
View File
@@ -1,7 +1,7 @@
import type { StoreApi } from 'zustand';
import type { AppState } from '@/lib/state/store';
import { registerAgentsHandlers } from '@/lib/ws/handlers/agents';
import { registerAgentSessionHandlers } from '@/lib/ws/handlers/agent-session';
import { registerTaskSessionHandlers } from '@/lib/ws/handlers/agent-session';
import { registerBoardsHandlers } from '@/lib/ws/handlers/boards';
import { registerColumnsHandlers } from '@/lib/ws/handlers/columns';
import { registerMessagesHandlers } from '@/lib/ws/handlers/messages';
@@ -26,7 +26,7 @@ export function registerWsHandlers(store: StoreApi<AppState>) {
...registerExecutorsHandlers(store),
...registerEnvironmentsHandlers(store),
...registerAgentsHandlers(store),
...registerAgentSessionHandlers(store),
...registerTaskSessionHandlers(store),
...registerUsersHandlers(store),
...registerTerminalsHandlers(store),
...registerDiffsHandlers(store),