53 lines
2.1 KiB
Go
53 lines
2.1 KiB
Go
package application
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"git.cin.su/m/Assistant/contracts/work"
|
|
"git.cin.su/m/Assistant/modules/work/domain"
|
|
"git.cin.su/m/Assistant/modules/work/ports"
|
|
)
|
|
|
|
func ExecuteAgentDispatch(ctx context.Context, store ports.ExecutionStore, executor ports.AgentExecutor, intent work.AgentDispatchIntent, workerID string, token work.AgentFencingToken, now time.Time, ttl time.Duration) (work.AgentExecutionResult, error) {
|
|
if executor == nil {
|
|
return work.AgentExecutionResult{}, ports.ErrAgentExecutorUnavailable
|
|
}
|
|
if err := domain.ValidateDispatchIntent(intent); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
lease, err := domain.AcquireAgentLease(intent.RunID, workerID, token, now, ttl)
|
|
if err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
if err := store.SaveDispatchIntent(ctx, intent); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
if err := store.SaveAgentLease(ctx, lease); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
heartbeat := work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: 1, Kind: work.AgentCallbackHeartbeat, ObservedAt: now}
|
|
if err := domain.ValidateAgentCallback(heartbeat, lease, 0); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
if err := store.AcceptAgentCallback(ctx, heartbeat); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
result, err := executor.Execute(ctx, work.AgentExecutionRequest{Intent: intent, WorkerID: workerID, FencingToken: token, LeaseExpiresAt: lease.ExpiresAt})
|
|
if err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
observedAt := time.UnixMilli(result.ObservedAtUnixMillis)
|
|
callback := work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: 2, Kind: work.AgentCallbackResult, ObservedAt: observedAt}
|
|
if result.State == work.AgentExecutionWaiting {
|
|
callback.Kind = work.AgentCallbackCheckpoint
|
|
}
|
|
if err := domain.ValidateAgentCallback(callback, lease, heartbeat.Sequence); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
if err := store.AcceptAgentCallback(ctx, callback); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
return result, nil
|
|
}
|