Interpreter Deep Dive

export function inferDependencies(step: LobsterXStep): string[] {
  const deps = new Set<string>(step.depends_on ?? []);

  if (step.stdin) {
    // Match "$step_id.stdout" or "$step_id.approved"
    const match = step.stdin.match(/^\$([a-z][a-z0-9_]*)\./);
    if (match?.[1]) {
      deps.add(match[1]);
    }
  }

  return Array.from(deps);
}

export function resolveExecutionLayers(steps: LobsterXStep[]): LobsterXStep[][] {
  const stepMap = new Map<string, LobsterXStep>();
  const depsMap = new Map<string, string[]>();

  for (const step of steps) {
    stepMap.set(step.id, step);
    depsMap.set(step.id, inferDependencies(step));
  }

  const layers: LobsterXStep[][] = [];
  const resolved = new Set<string>();
  const remaining = new Set(steps.map((s) => s.id));

  while (remaining.size > 0) {
    const layer: LobsterXStep[] = [];

    for (const id of remaining) {
      const deps = depsMap.get(id) ?? [];
      if (deps.every((d) => resolved.has(d))) {
        layer.push(stepMap.get(id)!);
      }
    }

    if (layer.length === 0) {
      throw new Error(
        `Circular or unresolvable dependencies detected among steps: ${Array.from(remaining).join(", ")}`
      );
    }

    layers.push(layer);
    for (const step of layer) {
      resolved.add(step.id);
      remaining.delete(step.id);
    }
  }

  return layers;
}

// convex/schema/step_results.ts
export const STEP_STATUSES = [
  "pending",
  "running",
  "completed",
  "failed",
  "skipped",
  "timed_out",
] as const;
export type StepStatus = (typeof STEP_STATUSES)[number];

while (true) {
  1. Find ready steps (deps met, not dispatched/completed/skipped/failed/cancelled)
  2. Dispatch each ready step (fire-and-forget)
  3. Check termination: all steps resolved? -> break
  4. awaitEvent("completion") -- single wait for any completion
  5. Route completion by type (step_complete | approval | subworkflow)
  6. Process result -> update state accumulators
  7. Loop back to 1
}

// convex/lib/interpreter/helpers.ts
export type CompletionEventType = "step_complete" | "approval" | "subworkflow";

export type StepCompleteEvent = {
  type: "step_complete";
  stepId: string;
  status?: string;
  outputs?: unknown;
  error?: string;
  startedAt?: number;
  completedAt?: number;
  model?: string;
  inputTokens?: number;
  outputTokens?: number;
  thinkingTokens?: number;
  totalTokens?: number;
  cost?: number;
  sessionId?: string;
  executionPath?: "convex-native" | "bridge-sync" | "bridge";
};

condition defined?
+-- No -> dispatch step normally
+-- Yes -> evaluateCondition(condition, accumulatedOutputs, variables)
    +-- Parse error -> fail step, apply on_failure policy
    +-- shouldRun: true -> dispatch step
    +-- shouldRun: false -> mark skipped, emit notification

export function resolveDotPath(
  path: string,
  outputs: Record<string, Record<string, unknown>>,
  variables: Record<string, unknown>,
): unknown {
  const parts = path.split(".");

  // Try step outputs first: "step_id.outputs.field" or "step_id.field"
  if (parts.length >= 2) {
    const stepId = parts[0]!;
    const stepOutputs = outputs[stepId];
    if (stepOutputs) {
      let current: unknown = stepOutputs;
      const fieldParts = parts[1] === "outputs" ? parts.slice(2) : parts.slice(1);
      for (const part of fieldParts) {
        if (current && typeof current === "object"
            && part in (current as Record<string, unknown>)) {
          current = (current as Record<string, unknown>)[part];
        } else {
          return undefined;
        }
      }
      return current;
    }
  }

  // Fall back to variables
  return variables[path] ?? variables[parts[0]!];
}

// convex/dispatch.ts
export class SafetyGateError extends Error {
  errorCode: string;
  retryable: boolean;

  constructor(errorCode: string, retryable: boolean, message: string) {
    super(message);
    this.name = "SafetyGateError";
    this.errorCode = errorCode;
    this.retryable = retryable;
  }
}

// From the .lobsterX schema (packages/shared/src/lobsterx-schema.ts)
on_failure: z.enum(["halt", "retry_once", "retry_once_then_escalate", "skip"]).optional()

export type PolicyResult = {
  result: "completed" | "skipped" | "retry" | "escalate" | "halted";
  error?: string;
};

parallel_failure_policy: z.enum(["wait_all", "fail_fast"]).optional()

{
  completedStepIds: Array.from(completedSteps),
  skippedStepIds: Array.from(skippedSteps),
  cancelledStepIds: Array.from(cancelledSteps),
  variableOverwriteOrder,
  variables,
  accumulatedOutputs,
}

success_criteria:
  - "outputs.risk_score >= 1"
  - "outputs.findings.length > 0"
  - "outputs.budget_impact in ['low','medium','high']"

// From completion_handlers.ts
const merged = { ...agentOutputs };
if (lobsterXStep.inputs) {
  for (const [key, inputDef] of Object.entries(lobsterXStep.inputs)) {
    if (lobsterXStep.outputs && key in lobsterXStep.outputs) continue; // agent owns this field
    // ... resolve from upstream outputs and carry forward
  }
}

export interface StepContext {
  runId: Id<"workflowRuns">;
  workflowName: string;
  workflowVersion: number;
  gatewayId: Id<"gatewayRegistry">;
  roleBindings: Record<string, string>;
  accumulatedOutputs: Record<string, Record<string, unknown>>;
  stepResults: Record<string, { outputs?: unknown; stdout?: string }>;
  skippedSteps: Set<string>;
  variables: Record<string, unknown>;
  definition: LobsterXDocument;
  variableOverwriteOrder: string[];
  traceId?: string;
  rootSpanId?: string;
}