popagent

Latest change 81b1cb620a6e0b4e696cfbf27e6bf93be875f4a9 - Mark the calling task in companyStatus so reports do not read themselves as a failing duplicate by AkurAI Build

import { createTool } from "@mastra/core/tools";
import { z } from "zod";
import type { AgentSchedule, AgentTask, AgentWorkspace, EvolutionSignal } from "../api-types";
import { AgentTaskStore } from "../tasks";
import { agentWorkspaces } from "../agent-workspaces";
import { evolutionStore } from "../evolution-store";
import { AGENT_CONTEXT_KEYS, agentContextValue } from "../agent-context";

const ATTENTION_STATUSES = new Set<AgentTask["status"]>(["failed", "cancelled", "dead-letter"]);
const EXCERPT = 400;

export type CompanyStatusSource = {
  workspaces(): Promise<AgentWorkspace[]>;
  tasks(workspaceId?: string): Promise<AgentTask[]>;
  schedules(workspaceId?: string): Promise<AgentSchedule[]>;
  automation(): Promise<{ state: string; selectedWorkspaceIds: string[]; activeTask: AgentTask | null }>;
  signals(limit: number): Promise<EvolutionSignal[]>;
};

const excerpt = (text: string | null | undefined, limit = EXCERPT) =>
  text ? (text.length > limit ? `${text.slice(0, limit)}…` : text) : null;

/**
 * Read-only snapshot of live Popagent state for reporting agents. Workspace
 * shell commands run in a network-disabled sandbox and cannot reach the
 * operator API, so this tool is the supported way to inspect the task board,
 * Needs-attention bucket, schedules, automation state, and recent signals.
 */
export function createCompanyStatusTool(source: CompanyStatusSource) {
  return createTool({
    id: "company-status",
    description: "Read-only snapshot of live Popagent state: registered workspaces, tasks since a time window (grouped by status, with outcomes), the Needs-attention bucket (failed/cancelled/dead-letter), schedules with their latest run, autonomy state, and recent evolution signals. Use this instead of shelling out to the popagent CLI or API, which are unreachable from the workspace sandbox. Output is bounded; ask for a narrower window or workspace if truncated.",
    inputSchema: z.object({
      sinceHours: z.number().int().min(1).max(168).default(24).describe("Look-back window for tasks and signals."),
      workspaceId: z.string().min(1).max(200).optional().describe("Limit tasks and schedules to one workspace id."),
      includeOutputs: z.boolean().default(true).describe("Include a bounded excerpt of each completed task's output."),
      maxTasks: z.number().int().min(1).max(200).default(60),
    }).strict(),
    outputSchema: z.object({
      generatedAt: z.string(),
      sinceHours: z.number(),
      workspaces: z.array(z.object({ id: z.string(), name: z.string(), repositoryPath: z.string() })),
      automation: z.object({ state: z.string(), selectedWorkspaceIds: z.array(z.string()), activeTaskId: z.string().nullable() }),
      taskCounts: z.record(z.string(), z.number()),
      tasks: z.array(z.record(z.string(), z.unknown())),
      needsAttention: z.array(z.record(z.string(), z.unknown())),
      schedules: z.array(z.record(z.string(), z.unknown())),
      signals: z.array(z.record(z.string(), z.unknown())),
      truncated: z.boolean(),
    }),
    execute: async ({ sinceHours, workspaceId, includeOutputs, maxTasks }, context) => {
      context.abortSignal?.throwIfAborted();
      // The calling task's own row would otherwise read as "another run of
      // this cycle, currently failing" — its error field is the previous
      // attempt's, and its status is this very run.
      const currentTaskId = agentContextValue(context.requestContext, AGENT_CONTEXT_KEYS.taskId);
      const since = Date.now() - sinceHours * 3_600_000;
      const [workspaces, allTasks, schedules, automation, signals] = await Promise.all([
        source.workspaces(),
        source.tasks(workspaceId),
        source.schedules(workspaceId),
        source.automation(),
        source.signals(50),
      ]);
      const recent = allTasks.filter((task) => new Date(task.completedAt ?? task.createdAt).getTime() >= since || task.status === "queued" || task.status === "running");
      const taskCounts: Record<string, number> = {};
      for (const task of recent) taskCounts[task.status] = (taskCounts[task.status] ?? 0) + 1;
      const shape = (task: AgentTask) => ({
        id: task.id,
        ...(task.id === currentTaskId ? { current: true, note: "This is the task you are executing right now; its error field belongs to a previous attempt." } : {}),
        workspaceId: task.workspaceId,
        source: task.source,
        scheduleId: task.scheduleId,
        status: task.status,
        model: task.model,
        attempts: `${task.attemptCount}/${task.maxAttempts}`,
        stepsCompleted: task.stepsCompleted,
        prompt: excerpt(task.prompt, 200),
        error: task.id === currentTaskId ? null : excerpt(task.error, 200),
        errorClass: task.id === currentTaskId ? null : task.lastErrorClass ?? null,
        createdAt: task.createdAt,
        completedAt: task.completedAt,
        output: includeOutputs && task.status === "completed" ? excerpt(task.output) : undefined,
      });
      const ordered = [...recent].sort((a, b) => (b.completedAt ?? b.createdAt).localeCompare(a.completedAt ?? a.createdAt));
      const needsAttention = allTasks.filter((task) => ATTENTION_STATUSES.has(task.status)).slice(0, 50).map(shape);
      return {
        generatedAt: new Date().toISOString(),
        sinceHours,
        workspaces: workspaces.map(({ id, name, repositoryPath }) => ({ id, name, repositoryPath })),
        automation: { state: automation.state, selectedWorkspaceIds: automation.selectedWorkspaceIds, activeTaskId: automation.activeTask?.id ?? null },
        taskCounts,
        tasks: ordered.slice(0, maxTasks).map(shape),
        needsAttention,
        schedules: schedules.map((schedule) => ({
          id: schedule.id,
          name: schedule.name,
          workspaceId: schedule.workspaceId,
          source: schedule.source,
          model: schedule.model,
          cron: schedule.cron,
          enabled: schedule.enabled,
          lastRunAt: schedule.lastRunAt,
          nextRunAt: schedule.nextRunAt,
          latestTaskStatus: schedule.latestTaskStatus ?? null,
        })),
        signals: signals.filter((signal) => new Date(signal.createdAt).getTime() >= since).slice(0, 30).map((signal) => ({
          id: signal.id,
          kind: signal.kind,
          status: signal.status,
          agentId: signal.agentId,
          workspaceId: signal.workspaceId,
          summary: excerpt(signal.summary, 200),
          createdAt: signal.createdAt,
        })),
        truncated: ordered.length > maxTasks,
      };
    },
  });
}

const store = new AgentTaskStore();
export const companyStatus = createCompanyStatusTool({
  workspaces: () => agentWorkspaces.list(),
  tasks: (workspaceId) => store.listTasks(workspaceId),
  schedules: (workspaceId) => store.listSchedules(workspaceId),
  automation: () => store.automationStatus(),
  signals: (limit) => evolutionStore.listSignals(limit),
});