Causation columns and activation probe
apps/backend/internal/office/repository/sqlite/base_migrations.go ↗Adds causation_id to three tables with partial indexes and publishes an activation instant only after all three columns exist.
Migration
func (r *Repository) migrateLoopLivenessCausationID() {
_ = r.migrate.Apply("office_routine_runs.causation_id",
`ALTER TABLE office_routine_runs ADD COLUMN causation_id TEXT NOT NULL DEFAULT ''`)
_ = r.migrate.Apply("idx_office_routine_runs_causation_id",
`CREATE INDEX IF NOT EXISTS idx_office_routine_runs_causation_id
ON office_routine_runs(causation_id) WHERE causation_id != ''`)
_ = r.migrate.Apply("agent_wakeup_requests.causation_id",
`ALTER TABLE agent_wakeup_requests ADD COLUMN causation_id TEXT NOT NULL DEFAULT ''`)
_ = r.migrate.Apply("idx_agent_wakeup_requests_causation_id",
`CREATE INDEX IF NOT EXISTS idx_agent_wakeup_requests_causation_id
ON agent_wakeup_requests(causation_id) WHERE causation_id != ''`)
_ = r.migrate.Apply("runs.causation_id",
`ALTER TABLE runs ADD COLUMN causation_id TEXT NOT NULL DEFAULT ''`)
_ = r.migrate.Apply("idx_runs_causation_id",
`CREATE INDEX IF NOT EXISTS idx_runs_causation_id
ON runs(causation_id) WHERE causation_id != ''`)
}
Activation
func (r *Repository) activateLoopLiveness() {
for _, probe := range []struct{ table, column string }{
{"office_routine_runs", causationIDColumn},
{"agent_wakeup_requests", causationIDColumn},
{"runs", causationIDColumn},
} {
exists, err := columnExists(r.db, probe.table, probe.column)
if err != nil || !exists {
return
}
}
persistence.WriteMetaKeyIfAbsent(r.db, loopLivenessActivationKey, time.Now().UTC().Format(time.RFC3339))
}
Routine fire recency and causation mint
apps/backend/internal/office/routines/service.go ↗Mints one causation id per fire, advances last_run_at before concurrency policy, and counts the fire by source and disposition.
Mint and persist
run := &RoutineRun{
RoutineID: routine.ID,
TriggerID: triggerID,
Source: source,
Status: models.RoutineRunStatusReceived,
StartedAt: &now,
CausationID: uuid.New().String(),
}
if err := s.repo.CreateRoutineRun(ctx, run); err != nil {
return nil, fmt.Errorf("create run: %w", err)
}
disposition := string(models.RoutineRunStatusFailed)
defer func() {
service.IncLoopRoutineRun(routine.WorkspaceID, source, disposition)
}()
if err := s.repo.TouchRoutineLastRun(ctx, routine.ID, *run.StartedAt); err != nil {
service.IncLoopLastRunAtWriteFailed(routine.WorkspaceID)
s.logger.Warn("touch routine last_run_at", zap.String("routine_id", routine.ID), zap.Error(err))
}
Monotonic touch
UPDATE office_routines
SET last_run_at = ?, updated_at = ?
WHERE id = ? AND (last_run_at IS NULL OR last_run_at < ?)
Wakeup adapter carries causation
apps/backend/internal/backendapp/adapters_office.go ↗Copies the fire's causation id onto the wakeup request so the lightweight path stays correlated.
Adapter
func (a *routineWakeupAdapter) CreateWakeupRequest(ctx context.Context, req *officeroutines.WakeupRequest) error {
row := &officesqlite.WakeupRequest{
ID: req.ID,
AgentProfileID: req.AgentProfileID,
Source: req.Source,
Reason: req.Reason,
Payload: req.Payload,
RequestedAt: req.RequestedAt,
CausationID: req.CausationID,
}
if err := a.repo.CreateWakeupRequest(ctx, row); err != nil {
if errors.Is(err, officesqlite.ErrWakeupIdempotencyConflict) {
return officeroutines.ErrWakeupAlreadyRequested
}
return err
}
return nil
}
Lightweight materialize
req := &WakeupRequest{
ID: uuid.New().String(),
AgentProfileID: routine.AssigneeAgentProfileID,
Source: "routine",
Payload: payloadStr,
RequestedAt: time.Now().UTC(),
CausationID: run.CausationID,
}
if err := s.wakeup.CreateWakeupRequest(ctx, req); err != nil {
return s.finalizeLightweightRun(ctx, run, models.RoutineRunStatusFailed)
}
service.IncLoopWakeupCreated(routine.WorkspaceID, req.Source)
Session hop persisted on both launch paths
apps/backend/internal/backendapp/main.go ↗Returns the orchestrator's session id and persists it on the runs row so a success claim carries evidence an agent ran.
Direct launch adapter
type officeOrchestratorTaskStarter struct { orch *orchestrator.Service }
var _ officeservice.TaskStarterWithLaunchContextSession = (*officeOrchestratorTaskStarter)(nil)
func (a *officeOrchestratorTaskStarter) StartTaskWithLaunchContextReturningSession(ctx context.Context, taskID, agentProfileID string, launch officeservice.LaunchContext) (string, error) {
execution, err := a.startTaskWithEnvAndSkills(ctx, taskID, agentProfileID, launch.ExecutorID, launch.ExecutorProfileID, launch.Priority, launch.Prompt, launch.WorkflowStepID, launch.PlanMode, launch.Attachments, launch.Env, launch.AdditionalSkillSlugs)
if err != nil || execution == nil {
return "", err
}
return execution.SessionID, nil
}
Routed launch adapter
var _ officescheduler.TaskStarterWithSession = (*schedulerTaskStarterAdapter)(nil)
func (a *schedulerTaskStarterAdapter) StartTaskWithRouteReturningSession(ctx context.Context, taskID, agentProfileID string, launch officescheduler.LaunchContext, route officescheduler.RouteOverride) (string, error) {
execution, err := a.startTaskWithRoute(ctx, taskID, agentProfileID, launch, route)
if err != nil || execution == nil {
return "", err
}
return execution.SessionID, nil
}
Persist with guard
UPDATE runs SET session_id = ? WHERE id = ? AND ? != ''
Terminal shape classification
apps/backend/internal/office/service/terminal_shapes.go ↗Maps every terminal run to one of seven shapes so silent success and unlaunched skip are distinct from real success.
Shapes
const (
ShapePreActivation TerminalShape = "pre_activation"
ShapeLaunchedCompleted TerminalShape = "launched_completed"
ShapeLaunchedFailed TerminalShape = "launched_failed"
ShapeSilentSuccess TerminalShape = "silent_success"
ShapeUnlaunchedSkipped TerminalShape = "unlaunched_skipped"
ShapeUnlaunchedFailed TerminalShape = "unlaunched_failed"
ShapeUnclassified TerminalShape = "unclassified"
)
Classifier
func ClassifyTerminalRun(status string, outcome *string, sessionID string, requestedAt time.Time, activationInstant time.Time, activationPublished bool) TerminalShape {
if !activationPublished || requestedAt.Before(activationInstant) {
return ShapePreActivation
}
outcomeVal := ""
if outcome != nil { outcomeVal = *outcome }
return classifyByOutcome(status, outcomeVal, sessionID != "")
}
func classifyByOutcome(status, outcomeVal string, launched bool) TerminalShape {
switch {
case status == "finished" && outcomeVal == "processed" && launched:
return ShapeLaunchedCompleted
case status == "finished" && outcomeVal == "processed" && !launched:
return ShapeSilentSuccess
case status == "finished" && !launched && skipOutcomes[outcomeVal]:
return ShapeUnlaunchedSkipped
case terminalFailedStatuses[status] && launched:
return ShapeLaunchedFailed
case terminalFailedStatuses[status] && !launched:
return ShapeUnlaunchedFailed
default:
return ShapeUnclassified
}
}
Health verdict and per-hop counters
apps/backend/internal/office/dashboard/loop_health.go ↗Evaluates one verdict from overdue, stuck, and silent-success evidence and exposes workspace-filtered counters.
Verdict precedence
func decideLoopHealthVerdict(triggerTotal, eligibleCount, stuckTotal, silentTotal int) string {
switch {
case triggerTotal > 0:
return VerdictDead
case stuckTotal > 0 || silentTotal > 0:
return VerdictDegraded
case eligibleCount == 0:
return VerdictNotArmed
default:
return VerdictHealthy
}
}
Counters
var (
loopCronTickTotal = expvar.NewMap("office_loop_cron_tick_total")
loopTriggerClaimedTotal = expvar.NewMap("office_loop_trigger_claimed_total")
loopRoutineRunTotal = expvar.NewMap("office_loop_routine_run_total")
loopWakeupCreatedTotal = expvar.NewMap("office_loop_wakeup_created_total")
loopRunClaimedTotal = expvar.NewMap("office_loop_run_claimed_total")
loopLaunchTotal = expvar.NewMap("office_loop_launch_total")
loopTerminalTotal = expvar.NewMap("office_loop_terminal_total")
loopCronTickAt = expvar.NewString("office_loop_cron_tick_at")
loopProcessStartedAt = expvar.NewString("office_loop_process_started_at")
)
func IncLoopCronTick(atRFC3339 string) { loopCronTickTotal.Add("", 1); loopCronTickAt.Set(atRFC3339) }
func RecordLoopProcessStarted(atRFC3339 string) { loopProcessStartedAt.Set(atRFC3339) }
Routes
func registerLoopHealthRoutes(api *gin.RouterGroup, h *Handler) {
api.GET("/workspaces/:wsId/loop-health", h.getLoopHealth)
api.GET("/workspaces/:wsId/loop-counters", h.getLoopCounters)
}
