Skip to content
Merged
4 changes: 2 additions & 2 deletions packages/api/routes/dockerRoutes.ts
Original file line number Diff line number Diff line change
Expand Up @@ -385,7 +385,7 @@ export async function stopTaskExecution(taskIdOrJobId: string, options: StopTask
* Returns the container ID when the container was stopped, null otherwise.
*/
async function stopRunningTaskContainer(taskId: string, state: TaskState, options: StopTaskExecutionOptions): Promise<string | null> {
const entry = state.history.find(h => h.state === 'claude_execution' && h.metadata?.containerId);
const entry = state.history.findLast(h => h.state === 'claude_execution' && h.metadata?.containerId);
const containerId = entry?.metadata?.containerId;
if (!containerId) {
console.log(`[stop-execution] No container ID found for task ${taskId}, relying on abort signal`);
Expand Down Expand Up @@ -425,7 +425,7 @@ export function createDockerRoutes(deps: DockerRoutesDeps) {
return;
}
const state = JSON.parse(stateData) as { history: Array<{ state: string; metadata?: { containerId?: string; containerName?: string } }> };
const entry = state.history.find(h => h.state === 'claude_execution' && h.metadata?.containerId);
const entry = state.history.findLast(h => h.state === 'claude_execution' && h.metadata?.containerId);
if (!entry?.metadata?.containerId) {
res.status(404).json({ error: 'No Docker container info available for this task' });
return;
Expand Down
681 changes: 681 additions & 0 deletions packages/api/routes/goalRoutes.ts

Large diffs are not rendered by default.

1 change: 1 addition & 0 deletions packages/api/routes/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -29,3 +29,4 @@ export { createUserRepoPreferencesRoutes } from './userRepoPreferencesRoutes.js'
export { createAgentRuntimeRoutes } from './agentRuntimeRoutes.js';
export { createNotificationRoutes } from './notificationRoutes.js';
export { createAdminRoutes } from './adminRoutes.js';
export { createGoalRoutes } from './goalRoutes.js';
35 changes: 33 additions & 2 deletions packages/api/routes/liveDetailsRoutes.ts
Original file line number Diff line number Diff line change
Expand Up @@ -263,7 +263,7 @@ async function loadStoredExecutionOutput(redisClient: RedisClientType, sessionId
const output = await fs.readFile(outputPath, 'utf8');
return parseStoredOutputContent(output);
}
async function parseActiveExecutionOutput(redisClient: RedisClientType, db: Knex, taskId: string): Promise<ConversationResult | null> {
async function parseActiveExecutionOutput(redisClient: RedisClientType, db: Knex, taskId: string): Promise<(ConversationResult & { nativeGoal?: ReturnType<typeof parseRedisOutput>['nativeGoal'] }) | null> {
const output = await redisClient.get(`agent:output:${taskId}`);
if (!output?.trim()) return null;
const executionStartTimestamp = await findExecutionStartTimestamp(redisClient, db, taskId);
Expand All @@ -279,7 +279,8 @@ async function parseActiveExecutionOutput(redisClient: RedisClientType, db: Knex
}) as unknown as Array<Record<string, unknown>>,
todos: redisParsed.todos,
currentTask: redisParsed.currentTask,
tokenUsage: redisParsed.tokenUsage
tokenUsage: redisParsed.tokenUsage,
nativeGoal: redisParsed.nativeGoal,
};
}
const parsedOutput = parseStoredOutputContent(output);
Expand All @@ -288,6 +289,36 @@ async function parseActiveExecutionOutput(redisClient: RedisClientType, db: Knex
? withStableResultEventIds(taskId, 'redis', executionStartTimestamp ?? taskId, result)
: null;
}

/** Provider-aware local projection shared by task details and goal summaries. */
export async function projectTaskLiveDetails(
redisClient: RedisClientType,
db: Knex,
taskId: string,
sessionId?: string | null,
): Promise<(ConversationResult & { nativeGoal?: ReturnType<typeof parseRedisOutput>['nativeGoal'] }) | null> {
const active = await parseActiveExecutionOutput(redisClient, db, taskId);
if (active) return active;
try {
const details = sessionId ? await parseExecutionDetailsFromDb(db, taskId, sessionId) : null;
if (details) return details;
const history = await db('task_history').where({ task_id: taskId })
.orderBy('timestamp', 'desc').limit(20).select('metadata');
const records = history.reverse().flatMap(entry => {
try {
const metadata = typeof entry.metadata === 'string' ? JSON.parse(entry.metadata) : entry.metadata;
return Array.isArray(metadata?.goalOutputRecords)
? metadata.goalOutputRecords.filter((value: unknown): value is string => typeof value === 'string')
: [];
} catch { return []; }
});
if (records.length === 0) return null;
const stored = parseStoredOutputContent(records.join('\n'));
return stored.parsed ?? stored.rawFallback;
} catch {
return null;
}
}
export function parseStoredOutputContent(output: string): ParsedStoredOutput {
if (!output.trim()) return { parsed: null, rawFallback: null, format: 'unknown' };
const format = detectStoredOutputFormat(output);
Expand Down
40 changes: 37 additions & 3 deletions packages/api/routes/llmLogsRoutes.ts
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,28 @@ function applyLlmLogFilters<T extends Knex.QueryBuilder>(query: T, filters: LlmL
return query;
}

function applyGoalOwnershipFilter<T extends Knex.QueryBuilder>(query: T, ownerId?: string): T {
return query.whereNotExists(function() {
this.select('*').from('goals').whereRaw('goals.current_task_id = llm_logs.task_id');
if (ownerId) this.whereNot('goals.owner_id', ownerId);
}) as T;
}

function buildLlmLogQueries(
db: Knex,
selectColumns: string[],
hasGoalsTable: boolean,
ownerId?: string,
): { baseQuery: Knex.QueryBuilder; countQuery: Knex.QueryBuilder } {
const baseQuery = db('llm_logs').select(...selectColumns);
const countQuery = db('llm_logs').count('* as count');
if (!hasGoalsTable) return { baseQuery, countQuery };
return {
baseQuery: applyGoalOwnershipFilter(baseQuery, ownerId),
countQuery: applyGoalOwnershipFilter(countQuery, ownerId),
};
}

interface UsageMetricRecordRow {
id: number;
llm_log_id: number;
Expand Down Expand Up @@ -244,6 +266,7 @@ export function createLlmLogsRoutes(deps: LlmLogsRoutesDeps) {
/** Process-wide cache for work-ref column existence. Only caches `true`
* so a process started before the migration will re-check until it lands. */
let hasWorkRefColumnsCache = false;
let hasGoalsTableCache = false;

async function checkWorkRefColumns(): Promise<boolean> {
if (hasWorkRefColumnsCache) return true;
Expand All @@ -256,6 +279,17 @@ export function createLlmLogsRoutes(deps: LlmLogsRoutesDeps) {
}
}

async function checkGoalsTable(): Promise<boolean> {
if (hasGoalsTableCache) return true;
try {
const result = await db.schema.hasTable('goals');
if (result) hasGoalsTableCache = true;
return result;
} catch {
return false;
}
}

async function getLlmLogs(req: Request, res: Response): Promise<void> {
try {
// Validate pagination parameters
Expand Down Expand Up @@ -310,6 +344,7 @@ export function createLlmLogsRoutes(deps: LlmLogsRoutesDeps) {

// Check if work-reference columns exist (cached after first successful check)
const hasWorkRefColumns = await checkWorkRefColumns();
const hasGoalsTable = await checkGoalsTable();

// Build and execute queries
const baseColumns = [
Expand All @@ -327,9 +362,8 @@ export function createLlmLogsRoutes(deps: LlmLogsRoutesDeps) {
? [...baseColumns, ...workRefColumns]
: baseColumns;

const baseQuery = db('llm_logs').select(...selectColumns);

const countQuery = db('llm_logs').count('* as count');
const ownerId = req.user?.id ? String(req.user.id) : undefined;
const { baseQuery, countQuery } = buildLlmLogQueries(db, selectColumns, hasGoalsTable, ownerId);

// If work_type filter is requested but the column doesn't exist, return empty results
if (!hasWorkRefColumns && workType) {
Expand Down
3 changes: 3 additions & 0 deletions packages/api/routes/taskHelpers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,9 @@ export async function getTasksFromDb(
`;

const baseQuery = db('tasks as t')
.where(function() {
this.whereNull('t.task_type').orWhereNot('t.task_type', 'goal');
})
.join(latestHistorySubquery, function() {
this.on('t.task_id', '=', 'h.task_id').andOn('h.rn', '=', db!.raw('?', [1]));
})
Expand Down
6 changes: 6 additions & 0 deletions packages/api/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ import {
createUserRepoPreferencesRoutes,
createAgentRuntimeRoutes, createNotificationRoutes,
createAdminRoutes,
createGoalRoutes,
createInstanceCatalogRoutes,
attachmentUpload
} from './routes/index.js';
Expand Down Expand Up @@ -277,8 +278,13 @@ function setupRoutes(): void {
const adminRoutes = createAdminRoutes();
const instanceCatalogRoutes = createInstanceCatalogRoutes();
const agentVersionRoutes = createAgentVersionRoutes();
const goalRoutes = createGoalRoutes({ db, taskQueue, redisClient });

app.use(['/api/task/:taskId', '/api/task/:taskId/*path', '/api/tasks/:taskId', '/api/execution/:sessionId', '/api/execution/:sessionId/*path', '/api/llm-metrics/:correlationId'], goalRoutes.requireGoalTaskOwnership);

const operationalRoutes: RouteEntry[] = [
['get', '/api/goals/capabilities', goalRoutes.capabilities], ['get', '/api/goals', goalRoutes.list], ['post', '/api/goals', goalRoutes.create], ['get', '/api/goals/:goalId', goalRoutes.get],
['post', '/api/goals/:goalId/pause', goalRoutes.pause], ['post', '/api/goals/:goalId/resume', goalRoutes.resume], ['post', '/api/goals/:goalId/cancel', goalRoutes.cancel], ['patch', '/api/goals/:goalId/model', goalRoutes.requestModel], ['post', '/api/goals/:goalId/input', goalRoutes.input],
['get', '/api/status', statusRoutes.getStatus], ['get', '/api/tasks', taskRoutes.getTasks], ['get', '/api/tasks/revert-preview', taskRoutes.getRevertPreview], ['post', '/api/tasks/revert', taskRoutes.revertChanges],
['post', '/api/tasks/:taskId/followup', taskRoutes.postFollowup], ...createTaskDeleteRouteEntries({ taskRoutes }), ['get', '/api/task/:taskId/history', taskHistoryRoutes.getTaskHistory], ['get', '/api/task/:taskId/live-details', liveDetailsRoutes.getLiveDetails],
['get', '/api/task/:taskId/file-changes', fileChangesRoutes.getFileChanges], ['get', '/api/queue/stats', queueRoutes.getQueueStats], ['get', '/api/activity', queueRoutes.getActivity], ['get', '/api/metrics', queueRoutes.getMetrics],
Expand Down
134 changes: 134 additions & 0 deletions packages/api/services/goalProjection.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
import type { Knex } from 'knex';
import type { RedisClientType } from 'redis';
import {
parseGoalArtifacts,
type GoalArtifactStats,
type GoalLaunchStrategy,
} from '@propr/core';
import { projectTaskLiveDetails } from '../routes/liveDetailsRoutes.js';

export interface GoalProjectionRow {
goal_id: string;
owner_id: string;
owner_login: string;
repository: string;
objective: string;
launch_strategy: GoalLaunchStrategy;
initial_prompt: string;
base_branch: string | null;
branch_name: string | null;
worktree_path: string | null;
agent_id: string;
agent_alias: string;
agent_type: string;
requested_model: string;
effective_model: string | null;
max_parallel_tasks: number | null;
ultrafix: number | boolean | null;
desired_state: 'running' | 'paused' | 'cancelled';
result_state: 'completed' | 'failed' | 'cancelled' | null;
current_task_id: string;
session_id: string | null;
conversation_id: string | null;
run_generation: number;
run_claim: string | null;
claimed_at: string | null;
active_turn_id: string | null;
pause_confirmed_at: string | null;
resume_requested: number | boolean;
final_pr_number: number | null;
final_pr_url: string | null;
artifact_refs: string | unknown[] | null;
artifact_stats: string | GoalArtifactStats | null;
artifacts_checked_at: string | null;
failure_reason: string | null;
create_idempotency_key: string | null;
create_idempotency_operation: string | null;
create_payload_hash: string | null;
control_generation: number;
control_ack_generation: number;
task_reconciled_at: string | null;
created_at: string;
updated_at: string;
started_at: string | null;
paused_at: string | null;
paused_ms: number;
completed_at: string | null;
}

function parseStats(value: GoalProjectionRow['artifact_stats']): GoalArtifactStats {
if (value && typeof value === 'object') return value;
if (typeof value === 'string') {
try { return JSON.parse(value) as GoalArtifactStats; } catch { /* use zero projection */ }
}
return { issues: 0, openIssues: 0, pullRequests: 0, openPullRequests: 0 };
}

function goalTiming(row: GoalProjectionRow): { elapsedMs: number; pausedMs: number; activeMs: number } {
const endMs = row.completed_at ? new Date(row.completed_at).getTime() : Date.now();
const startMs = row.started_at ? new Date(row.started_at).getTime() : new Date(row.created_at).getTime();
const currentPauseMs = row.desired_state === 'paused' && row.paused_at
? Math.max(0, Date.now() - new Date(row.paused_at).getTime())
: 0;
const pausedMs = Number(row.paused_ms || 0) + currentPauseMs;
const elapsedMs = Math.max(0, endMs - startMs);
return { elapsedMs, pausedMs, activeMs: Math.max(0, elapsedMs - pausedMs) };
}

export async function serializeGoal(
db: Knex,
redis: RedisClientType,
source: GoalProjectionRow,
) {
const row = source;
const live = await projectTaskLiveDetails(redis, db, row.current_task_id, row.session_id);
const latestHistory = await db('task_history')
.where({ task_id: row.current_task_id })
.orderBy('timestamp', 'desc')
.first();
const timing = goalTiming(row);
return {
id: row.goal_id,
owner: row.owner_login,
repository: row.repository,
objective: row.objective,
launchStrategy: row.launch_strategy,
initialPrompt: row.initial_prompt,
baseBranch: row.base_branch,
branchName: row.branch_name,
worktreePath: row.worktree_path,
agent: { id: row.agent_id, alias: row.agent_alias, type: row.agent_type },
requestedModel: row.requested_model,
effectiveModel: row.effective_model,
maxParallelTasks: row.max_parallel_tasks,
ultrafix: row.ultrafix == null ? null : Boolean(row.ultrafix),
desiredState: row.desired_state,
resultState: row.result_state,
failureReason: row.failure_reason,
pausePending: row.desired_state === 'paused' && !row.pause_confirmed_at,
control: {
requestGeneration: Number(row.control_generation || 0),
acknowledgedGeneration: Number(row.control_ack_generation || 0),
pending: Number(row.control_ack_generation || 0) < Number(row.control_generation || 0),
},
taskId: row.current_task_id,
sessionId: row.session_id,
conversationId: row.conversation_id,
finalPr: row.final_pr_url ? { number: row.final_pr_number, url: row.final_pr_url } : null,
artifacts: parseGoalArtifacts(row.artifact_refs as string | null),
artifactStats: parseStats(row.artifact_stats),
liveSummary: {
currentTask: live?.currentTask ?? null,
todos: live?.todos ?? [],
tokenUsage: live?.tokenUsage ?? null,
nativeGoal: live?.nativeGoal ?? null,
},
taskState: latestHistory?.state ?? 'pending',
createdAt: row.created_at,
updatedAt: row.updated_at,
startedAt: row.started_at,
pausedAt: row.paused_at,
completedAt: row.completed_at,
...timing,
};
}
Loading
Loading