///|
///| Durable state projection for a MoonClaw-owned job.
///|
///| The pack derives state from plans and completed bundles only; it does not run
///|
/// an agent, invoke a model, or schedule retries itself.
pub(all) enum AnalysisJobOutcome {
ReadyToStart
AwaitingMarketData
Completed
Cancelled
Failed
} derive(Debug, Eq, ToJson, FromJson)
///|
pub(all) enum JobFailureKind {
PlanningFailed
MarketDataUnavailable
ModelGatewayUnavailable
ValidationFailed
RuntimeFailed
OperatorCancelled
} derive(Debug, Eq, ToJson, FromJson)
///|
pub(all) struct RetryPolicy {
max_attempts : Int
semantic_max_attempts : Int
retry_stage2 : Bool
} derive(Debug, Eq, ToJson, FromJson)
///|
pub(all) struct AnalysisJobInput {
plan : @planning.AnalysisPlan
bundle : @workflow.AnalysisRunBundle?
cancel_reason : String?
failure_message : String?
retry_policy : RetryPolicy
} derive(Debug, Eq, ToJson, FromJson)
///|
pub(all) struct AnalysisJobState {
run_id : @domain.RunId
outcome : AnalysisJobOutcome
failure_kind : JobFailureKind?
bundle_status : @workflow.AnalysisRunStatus?
can_start : Bool
can_retry : Bool
partial_record_available : Bool
retry_policy : RetryPolicy
event_count : Int
review_count : Int
blocking_review_count : Int
artifact_paths : Array[String]
events : Array[@routine.RoutineEvent]
message : String
} derive(Debug, Eq, ToJson, FromJson)
///|
pub fn AnalysisJobOutcome::label(self : AnalysisJobOutcome) -> String {
match self {
ReadyToStart => "ready_to_start"
AwaitingMarketData => "awaiting_market_data"
Completed => "completed"
Cancelled => "cancelled"
Failed => "failed"
}
}
///|
pub fn default_retry_policy(
max_attempts? : Int = 3,
semantic_max_attempts? : Int = 1,
retry_stage2? : Bool = true,
) -> RetryPolicy {
{ max_attempts, semantic_max_attempts, retry_stage2 }
}
///|
pub fn analysis_job_input(
plan : @planning.AnalysisPlan,
bundle? : @workflow.AnalysisRunBundle,
cancel_reason? : String,
failure_message? : String,
retry_policy? : RetryPolicy = default_retry_policy(),
) -> AnalysisJobInput {
{ plan, bundle, cancel_reason, failure_message, retry_policy }
}
///|
fn blocking_review_count(queue : @review.ReviewQueue) -> Int {
queue.items.fold(init=0, fn(count, item) {
if item.priority is Blocking {
count + 1
} else {
count
}
})
}
///|
fn bundle_events(
bundle : @workflow.AnalysisRunBundle,
) -> Array[@routine.RoutineEvent] {
match bundle.projection {
Some(projection) => projection.result.events
None => []
}
}
///|
fn bundle_artifact_paths(bundle : @workflow.AnalysisRunBundle) -> Array[String] {
match bundle.write_plan {
Some(plan) => plan.artifacts.map(fn(artifact) { artifact.path })
None => []
}
}
///|
fn append_terminal_event(
events : Array[@routine.RoutineEvent],
run_id : @domain.RunId,
kind : @routine.RoutineEventKind,
message : String,
) -> Array[@routine.RoutineEvent] {
@routine.append_event(
events,
@routine.routine_event(
run_id,
events.length(),
CreateRunContext,
kind,
message,
),
)
}
///|
fn planning_failure_kind(plan : @planning.AnalysisPlan) -> JobFailureKind {
if plan.gateway.health is Unavailable {
ModelGatewayUnavailable
} else {
PlanningFailed
}
}
///|
fn can_retry_failure(kind : JobFailureKind?, policy : RetryPolicy) -> Bool {
match kind {
Some(OperatorCancelled) => false
Some(PlanningFailed) => false
Some(MarketDataUnavailable) => false
Some(_) => policy.max_attempts > 0
None => false
}
}
///|
fn base_message(
outcome : AnalysisJobOutcome,
plan : @planning.AnalysisPlan,
) -> String {
match outcome {
ReadyToStart => "job is ready to start: \{plan.summary}"
AwaitingMarketData => "job is waiting for external market evidence"
Completed => "job completed and produced durable run artifacts"
Cancelled => "job was cancelled by operator request"
Failed => "job failed before completion"
}
}
///|
fn state(
plan : @planning.AnalysisPlan,
outcome : AnalysisJobOutcome,
failure_kind : JobFailureKind?,
bundle_status : @workflow.AnalysisRunStatus?,
retry_policy : RetryPolicy,
events : Array[@routine.RoutineEvent],
review_count : Int,
blocking_review_count : Int,
artifact_paths : Array[String],
message : String,
) -> AnalysisJobState {
{
run_id: plan.run_id,
outcome,
failure_kind,
bundle_status,
can_start: outcome is ReadyToStart || outcome is Completed,
can_retry: can_retry_failure(failure_kind, retry_policy),
partial_record_available: artifact_paths.length() > 0,
retry_policy,
event_count: events.length(),
review_count,
blocking_review_count,
artifact_paths,
events,
message,
}
}
///|
pub fn prepare_analysis_job(input : AnalysisJobInput) -> AnalysisJobState {
let plan = input.plan
let retry_policy = input.retry_policy
match input.cancel_reason {
Some(reason) => {
let events = append_terminal_event(
[],
plan.run_id,
RoutineCancelled,
"cancelled: \{reason}",
)
state(
plan,
Cancelled,
Some(OperatorCancelled),
None,
retry_policy,
events,
0,
0,
[],
"job cancelled before execution: \{reason}",
)
}
None =>
match input.failure_message {
Some(message) => {
let events = append_terminal_event(
[],
plan.run_id,
RoutineFailed,
"failed: \{message}",
)
state(
plan,
Failed,
Some(RuntimeFailed),
None,
retry_policy,
events,
0,
0,
[],
message,
)
}
None =>
if !plan.can_start {
let kind = planning_failure_kind(plan)
let events = append_terminal_event(
[],
plan.run_id,
RoutineFailed,
"planning failed: \{plan.summary}",
)
state(
plan,
Failed,
Some(kind),
None,
retry_policy,
events,
plan.issues.length(),
plan.issues.length(),
[],
"planning failed with \{plan.issues.length()} issue(s)",
)
} else {
match input.bundle {
None =>
state(
plan,
ReadyToStart,
None,
None,
retry_policy,
[],
0,
0,
[],
base_message(ReadyToStart, plan),
)
Some(bundle) => {
let queue = @review.review_queue_for_bundle(bundle)
let events = bundle_events(bundle)
let artifacts = bundle_artifact_paths(bundle)
let outcome = if bundle.status is Completed {
Completed
} else {
AwaitingMarketData
}
let kind = if outcome is AwaitingMarketData {
Some(MarketDataUnavailable)
} else {
None
}
state(
plan,
outcome,
kind,
Some(bundle.status),
retry_policy,
events,
queue.items.length(),
blocking_review_count(queue),
artifacts,
base_message(outcome, plan),
)
}
}
}
}
}
}