feat(orchestrator): MS23 agent lifecycle ingestion service (#701)
Some checks failed
ci/woodpecker/push/ci Pipeline failed
Some checks failed
ci/woodpecker/push/ci Pipeline failed
Co-authored-by: Jason Woltje <jason@diversecanvas.com> Co-committed-by: Jason Woltje <jason@diversecanvas.com>
This commit was merged in pull request #701.
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
import { Injectable, Logger, Inject, forwardRef } from "@nestjs/common";
|
||||
import { Injectable, Logger, Inject, Optional, forwardRef } from "@nestjs/common";
|
||||
import { ValkeyService } from "../valkey/valkey.service";
|
||||
import { AgentSpawnerService } from "./agent-spawner.service";
|
||||
import { AgentIngestionService } from "../agent-ingestion/agent-ingestion.service";
|
||||
import type { AgentState, AgentStatus, AgentEvent } from "../valkey/types";
|
||||
import { isValidAgentTransition } from "../valkey/types/state.types";
|
||||
|
||||
@@ -32,7 +33,8 @@ export class AgentLifecycleService {
|
||||
constructor(
|
||||
private readonly valkeyService: ValkeyService,
|
||||
@Inject(forwardRef(() => AgentSpawnerService))
|
||||
private readonly spawnerService: AgentSpawnerService
|
||||
private readonly spawnerService: AgentSpawnerService,
|
||||
@Optional() private readonly agentIngestionService?: AgentIngestionService
|
||||
) {
|
||||
this.logger.log("AgentLifecycleService initialized");
|
||||
}
|
||||
@@ -55,6 +57,25 @@ export class AgentLifecycleService {
|
||||
return createdState;
|
||||
}
|
||||
|
||||
private async recordLifecycleIngestion(
|
||||
agentId: string,
|
||||
event: "started" | "completed" | "failed" | "killed",
|
||||
record: (ingestionService: AgentIngestionService) => Promise<void>
|
||||
): Promise<void> {
|
||||
if (!this.agentIngestionService) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await record(this.agentIngestionService);
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
this.logger.error(
|
||||
`Failed to record agent ${event} ingestion for ${agentId}: ${errorMessage}`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Acquire a per-agent mutex to serialize state transitions.
|
||||
* Uses promise chaining: each caller chains onto the previous lock,
|
||||
@@ -118,6 +139,10 @@ export class AgentLifecycleService {
|
||||
// Emit event
|
||||
await this.publishStateChangeEvent("agent.running", updatedState);
|
||||
|
||||
await this.recordLifecycleIngestion(agentId, "started", (ingestionService) =>
|
||||
ingestionService.recordAgentStarted(agentId)
|
||||
);
|
||||
|
||||
this.logger.log(`Agent ${agentId} transitioned to running`);
|
||||
return updatedState;
|
||||
});
|
||||
@@ -155,6 +180,10 @@ export class AgentLifecycleService {
|
||||
// Emit event
|
||||
await this.publishStateChangeEvent("agent.completed", updatedState);
|
||||
|
||||
await this.recordLifecycleIngestion(agentId, "completed", (ingestionService) =>
|
||||
ingestionService.recordAgentCompleted(agentId)
|
||||
);
|
||||
|
||||
// Schedule session cleanup
|
||||
this.spawnerService.scheduleSessionCleanup(agentId);
|
||||
|
||||
@@ -192,6 +221,10 @@ export class AgentLifecycleService {
|
||||
// Emit event
|
||||
await this.publishStateChangeEvent("agent.failed", updatedState, error);
|
||||
|
||||
await this.recordLifecycleIngestion(agentId, "failed", (ingestionService) =>
|
||||
ingestionService.recordAgentFailed(agentId, error)
|
||||
);
|
||||
|
||||
// Schedule session cleanup
|
||||
this.spawnerService.scheduleSessionCleanup(agentId);
|
||||
|
||||
@@ -228,6 +261,10 @@ export class AgentLifecycleService {
|
||||
// Emit event
|
||||
await this.publishStateChangeEvent("agent.killed", updatedState);
|
||||
|
||||
await this.recordLifecycleIngestion(agentId, "killed", (ingestionService) =>
|
||||
ingestionService.recordAgentKilled(agentId)
|
||||
);
|
||||
|
||||
// Schedule session cleanup
|
||||
this.spawnerService.scheduleSessionCleanup(agentId);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user