internal/app/application_lifecycle.go
Rama main · réplica de lectura del repositorio original.
package app
import (
"context"
"errors"
"fmt"
"net"
"net/http"
"net/url"
"time"
"github.com/yeomyeonggeori/blueclaw/internal/scheduler"
"github.com/yeomyeonggeori/blueclaw/internal/task"
"github.com/yeomyeonggeori/blueprotocol/agentcontract"
)
func (application *Application) Start() error {
if application.startupError != nil {
return application.startupError
}
if application.languageModelError != nil {
application.runtimeLogger.Logger.Error("application.language_model_configuration_rejected", "error", application.languageModelError.Error())
return application.serveWithoutStartingWork()
}
if errorValue := application.checkProtocolIdentity(); errorValue != nil {
application.runtimeLogger.Logger.Error("application.protocol_identity_rejected", "error", errorValue.Error())
return application.serveWithoutStartingWork()
}
application.startClockWatch()
if application.refreshSkillIndex != nil {
go application.refreshSkillIndex(context.Background())
}
application.startLearningCoordinator()
application.runtimeLogger.Logger.Info("application.starting", "stage", "log_retention")
application.startLogRetentionLoop()
application.runtimeLogger.Logger.Info("application.starting", "stage", "memory_carry_forward")
if errorValue := application.carryMemoryForward(); errorValue != nil {
return errorValue
}
application.runtimeLogger.Logger.Info("application.starting", "stage", "memory_worker")
application.startMemoryMaintenance()
application.runtimeLogger.Logger.Info("application.starting", "stage", "connector_runtime")
application.startConnectorRuntime()
application.runtimeLogger.Logger.Info("application.starting", "stage", "acp_session")
if errorValue := application.startACPSessionServer(); errorValue != nil {
return errorValue
}
application.runtimeLogger.Logger.Info("application.starting", "stage", "task_schedule")
application.startSchedulePoller()
application.runtimeLogger.Logger.Info("application.starting", "stage", "task_retention")
application.startTaskRetentionSweeper()
application.runtimeLogger.Logger.Info("application.starting", "stage", "stale_tasks")
application.startStaleTaskSweeper()
application.runtimeLogger.Logger.Info("application.starting", "stage", "listen")
listener, errorValue := net.Listen("tcp", application.httpServer.Addr)
if errorValue != nil {
return errorValue
}
application.runtimeLogger.Logger.Info(
"application.started",
"listenAddress",
application.httpServer.Addr,
"languageModelConfigured",
application.languageModelConfigured,
"logDirectoryPath",
application.runtimeLogger.DirectoryPath(),
)
application.startInterruptedTaskAutoResume()
return application.httpServer.Serve(listener)
}
// A contract this build does not share would have the agent call tool names
// the other side never registered, so no work may run. Exiting instead leaves
// the supervisor restarting every few seconds, which also takes down the one
// endpoint where the disagreement is visible to the host.
func (application *Application) serveWithoutStartingWork() error {
listener, errorValue := net.Listen("tcp", application.httpServer.Addr)
if errorValue != nil {
return errorValue
}
application.runtimeLogger.Logger.Info(
"application.serving_health_only",
"listenAddress",
application.httpServer.Addr,
)
return application.httpServer.Serve(listener)
}
// Handler exposes the built HTTP surface so a caller can exercise the runtime
// without binding a port.
func (application *Application) Handler() http.Handler {
if application.httpServer == nil {
return nil
}
return application.httpServer.Handler
}
func (application *Application) Shutdown(ctx context.Context) error {
if application.connectorRuntimeCancel != nil {
application.connectorRuntimeCancel()
}
if application.acpSessionCancel != nil {
application.acpSessionCancel()
}
if application.scheduleCancel != nil {
application.scheduleCancel()
}
if application.taskRetentionCancel != nil {
application.taskRetentionCancel()
}
if application.staleTaskCancel != nil {
application.staleTaskCancel()
}
if application.interruptedTaskResumeCancel != nil {
application.interruptedTaskResumeCancel()
}
if application.logRetentionCancel != nil {
application.logRetentionCancel()
}
if application.memoryMaintenanceCancel != nil {
application.memoryMaintenanceCancel()
}
if application.learningCancel != nil {
application.learningCancel()
}
if application.clockWatchCancel != nil {
application.clockWatchCancel()
}
errorValue := application.httpServer.Shutdown(ctx)
backgroundError := application.awaitBackgroundLoops(ctx)
terminalCloseError := application.closeTerminalSessions()
closeErrorValue := application.runtimeLogger.Close()
databaseCloseError := application.database.Close()
if errorValue != nil {
return errorValue
}
if backgroundError != nil {
return backgroundError
}
if terminalCloseError != nil {
return terminalCloseError
}
if closeErrorValue != nil {
return closeErrorValue
}
return databaseCloseError
}
func (application *Application) startLearningCoordinator() {
if application.learningCoordinator == nil || application.learningCancel != nil {
return
}
application.learningCancel = application.startBackgroundLoop(application.learningCoordinator.Run)
}
// Cancelling a context asks a goroutine to stop. Nothing was waiting for one to have
// stopped, so Shutdown closed the database while sweepers were still writing to it.
func (application *Application) startBackgroundLoop(run func(context.Context)) context.CancelFunc {
ctx, cancel := context.WithCancel(context.Background())
application.backgroundLoops.Add(1)
go func() {
defer application.backgroundLoops.Done()
run(ctx)
}()
return cancel
}
func (application *Application) awaitBackgroundLoops(ctx context.Context) error {
stopped := make(chan struct{})
go func() {
application.backgroundLoops.Wait()
close(stopped)
}()
select {
case <-stopped:
return nil
case <-ctx.Done():
return errors.New("shutdown ran out of time before its background loops stopped")
case <-time.After(backgroundLoopStopGrace):
return errors.New("background loops did not stop within " + backgroundLoopStopGrace.String())
}
}
func (application *Application) closeTerminalSessions() error {
if application.terminalService == nil {
return nil
}
return application.terminalService.CloseAllSessions()
}
func (application *Application) startConnectorRuntime() {
if application.connectorRuntime == nil || application.connectorRuntimeCancel != nil {
return
}
ctx, cancel := context.WithCancel(context.Background())
application.connectorRuntimeCancel = cancel
application.connectorRuntime.Start(ctx)
}
func (application *Application) startACPSessionServer() error {
if application.acpSessionServer == nil || application.acpSessionCancel != nil {
return nil
}
if errorValue := application.acpSessionServer.Listen(); errorValue != nil {
return errorValue
}
ctx, cancel := context.WithCancel(context.Background())
application.acpSessionCancel = cancel
go application.acpSessionServer.Serve(ctx)
return nil
}
func (application *Application) startLogRetentionLoop() {
if application.runtimeLogger == nil || application.logRetentionCancel != nil {
return
}
application.logRetentionCancel = application.startBackgroundLoop(application.runtimeLogger.StartRetentionLoop)
}
func (application *Application) startSchedulePoller() {
if application.schedulePoller == nil || application.scheduleCancel != nil {
return
}
interval := time.Duration(application.schedulePollIntervalSecond()) * time.Second
application.scheduleCancel = application.startBackgroundLoop(func(ctx context.Context) {
application.schedulePoller.Start(ctx, interval)
})
}
func (application *Application) startTaskRetentionSweeper() {
if application.taskRetentionSweeper == nil || application.taskRetentionCancel != nil {
return
}
interval := time.Duration(application.taskRetentionIntervalMinuteOrDefault()) * time.Minute
application.taskRetentionCancel = application.startOnceTheClockIsTrusted(func(ctx context.Context) {
application.taskRetentionSweeper.Start(ctx, interval)
})
}
func (application *Application) startStaleTaskSweeper() {
if application.taskRunService == nil || application.staleTaskCancel != nil {
return
}
sweeper := scheduler.StaleTaskSweeper{
TaskRunService: application.taskRunService,
Notifier: application.interruptedTaskResumer,
Logger: application.runtimeLogger.Logger,
}
application.staleTaskCancel = application.startOnceTheClockIsTrusted(func(ctx context.Context) {
sweeper.Start(ctx, 30*time.Minute)
})
}
func (application *Application) startInterruptedTaskAutoResume() {
if application.taskRunService == nil || application.interruptedTaskResumer == nil || application.interruptedTaskResumeCancel != nil {
return
}
application.interruptedTaskResumeCancel = application.startOnceTheClockIsTrusted(func(ctx context.Context) {
application.resumeInterruptedTaskRuns(ctx, time.Now())
})
}
func (application *Application) resumeInterruptedTaskRuns(ctx context.Context, now time.Time) {
selection := application.taskRunService.SelectInterruptedTaskRunsForAutoResume(now, 5)
for _, taskRun := range selection.SkippedTaskRuns {
application.taskRunService.MarkInterruptedTaskRunAutoResumeSkipped(taskRun.TaskRunID, "per_boot_limit_exceeded")
}
for index, taskRun := range selection.SelectedTaskRuns {
if ctx.Err() != nil {
return
}
if index > 0 && !application.waitBeforeInterruptedTaskResume(ctx) {
return
}
if !application.interruptedTaskResumer.CanResumeInterruptedTaskRun(taskRun) {
application.taskRunService.MarkInterruptedTaskRunAutoResumeSkipped(taskRun.TaskRunID, "resume_context_unavailable")
continue
}
if !application.taskRunService.ClaimInterruptedTaskRunAutoResume(taskRun.TaskRunID, "runtime_restart") {
continue
}
if _, errorValue := application.interruptedTaskResumer.ResumeInterruptedTaskRun(ctx, taskRun); errorValue != nil {
application.taskRunService.AppendTaskEvent(taskRun.TaskRunID, agentcontract.TaskEventTaskAutoResumeLaunchFailed, errorValue.Error())
}
}
application.failUnresumedInterruptedTaskRuns(ctx)
}
func (application *Application) failUnresumedInterruptedTaskRuns(ctx context.Context) {
for _, taskRun := range application.taskRunService.ListTaskRun() {
if ctx.Err() != nil {
return
}
if !task.TaskRunWasInterruptedByRuntimeRestart(taskRun) {
continue
}
application.taskRunService.AppendTaskEvent(taskRun.TaskRunID, agentcontract.TaskEventTaskAutoResumeAbandoned, taskRun.FailureReason)
application.interruptedTaskResumer.FailUnresumedInterruptedTaskRun(ctx, taskRun, "the task was interrupted by a runtime restart and could not be resumed")
}
}
func (application *Application) waitBeforeInterruptedTaskResume(ctx context.Context) bool {
delay := application.interruptedTaskResumeDelay
if delay <= 0 {
return true
}
timer := time.NewTimer(delay)
defer timer.Stop()
select {
case <-ctx.Done():
return false
case <-timer.C:
return true
}
}
func (application *Application) taskRetentionIntervalMinuteOrDefault() int {
if application.taskRetentionIntervalMinute > 0 {
return application.taskRetentionIntervalMinute
}
return 60
}
// carryMemoryForward brings memory written under the layout that kept every
// subject's file in one service-owned directory into the file inside each
// subject's own workspace directory. Nothing else reads the old path, so a
// start that carried nothing would answer from an empty memory without
// saying so, which is why a failure here stops the start instead of being
// logged and stepped over.
func (application *Application) carryMemoryForward() error {
if application.memoryStores == nil {
return nil
}
report, errorValue := application.memoryStores.CarryForward(context.Background())
if errorValue != nil {
return fmt.Errorf("carry the memory written at the old path: %w", errorValue)
}
if report.Carried > 0 || report.Adopted > 0 {
application.runtimeLogger.Logger.Info("application.memory.carried_forward", "carried", report.Carried, "adopted", report.Adopted)
}
if len(report.Left) > 0 {
application.runtimeLogger.Logger.Warn("application.memory.left_at_old_path", "files", report.Left)
}
if len(report.Removed) > 0 {
application.runtimeLogger.Logger.Info("application.memory.placeless_circle_removed", "folders", report.Removed)
}
if errorValue := application.carryMarkdownMemory(); errorValue != nil {
return errorValue
}
return application.removeRetiredCircleFolders()
}
func (application *Application) carryMarkdownMemory() error {
report, errorValue := application.memoryStores.CarryMarkdownMemory(context.Background(), application.isOnTheRoster)
if errorValue != nil {
return fmt.Errorf("carry the memory written as markdown: %w", errorValue)
}
if report.Carried > 0 {
application.runtimeLogger.Logger.Info("application.memory.markdown_carried", "people", report.Carried)
}
if len(report.Left) > 0 {
application.runtimeLogger.Logger.Warn("application.memory.markdown_left", "files", report.Left)
}
return nil
}
// retiredCircleIDs are the circles internkim made by default, or hosts were
// found carrying, before a circle became the data room role a person holds.
var retiredCircleIDs = []string{"c-level", "representative", "hr", "staff", "hr-compensation"}
func (application *Application) removeRetiredCircleFolders() error {
held := application.heldCircles()
unheld := []string{}
for _, circleID := range retiredCircleIDs {
if !held[circleID] {
unheld = append(unheld, circleID)
}
}
report, errorValue := application.memoryStores.RemoveRetiredCircleFolders(context.Background(), unheld)
if errorValue != nil {
return fmt.Errorf("remove the folders of circles nobody holds: %w", errorValue)
}
if len(report.Removed) > 0 {
application.runtimeLogger.Logger.Info("application.memory.retired_circle_removed", "folders", report.Removed)
}
if len(report.Left) > 0 {
application.runtimeLogger.Logger.Warn("application.memory.retired_circle_kept", "folders", report.Left)
}
return nil
}
func (application *Application) startMemoryMaintenance() {
if application.memoryStores == nil || application.memoryMaintenanceCancel != nil {
return
}
ctx, cancel := context.WithCancel(context.Background())
application.memoryMaintenanceCancel = cancel
go application.maintainMemory(ctx)
}
func (application *Application) maintainMemory(ctx context.Context) {
for {
waitDuration := memoryMaintenanceInterval
if !application.runMemoryMaintenance(ctx) {
waitDuration = memoryMaintenanceRetryInterval
}
select {
case <-ctx.Done():
return
case <-time.After(waitDuration):
}
}
}
func (application *Application) runMemoryMaintenance(ctx context.Context) bool {
report, errorValue := application.memoryStores.Maintain(ctx)
if ctx.Err() != nil {
return true
}
logger := application.runtimeLogger.Logger
if errorValue != nil {
logger.Warn("application.memory.maintenance_failed", "error", errorValue.Error(),
"scopes", report.Scopes, "reembeddedMemories", report.Reembedded.Memories)
return false
}
logger.Info("application.memory.maintenance_completed",
"scopes", report.Scopes,
"reembeddedMemories", report.Reembedded.Memories,
"reembeddedTriggers", report.Reembedded.Triggers,
"reembeddedFiles", report.Reembedded.Files)
return true
}
func (application *Application) schedulePollIntervalSecond() int {
if application.schedulePollSecond > 0 {
return application.schedulePollSecond
}
return 30
}
func deriveListenAddress(baseURL string) string {
if baseURL == "" {
return "127.0.0.1:8080"
}
parsedURL, errorValue := url.Parse(baseURL)
if errorValue != nil || parsedURL.Host == "" {
return baseURL
}
return parsedURL.Host
}