diff --git a/clients/swift/FractalClient/Sources/FractalClient/Models.swift b/clients/swift/FractalClient/Sources/FractalClient/Models.swift index 4b82447846e..cef10d7387b 100644 --- a/clients/swift/FractalClient/Sources/FractalClient/Models.swift +++ b/clients/swift/FractalClient/Sources/FractalClient/Models.swift @@ -615,7 +615,9 @@ public struct AgentPresentation: Codable, Sendable { public let state, group: St public struct AgentPresentationWork: Codable, Sendable { public let step, mission, path, label: String; public let title: String? } public struct AgentPresentationNow: Codable, Sendable { public let step, mission, path, label, since: String; public let title, goal, holderDownSince: String?; enum CodingKeys: String, CodingKey { case step, mission, path, label, since, title, goal, holderDownSince = "holder_down_since" } } public struct WorkLabel: Codable, Sendable { public let id, missionID, missionRunID, path, state, since: String; public let title, goal, holderDownSince: String?; enum CodingKeys: String, CodingKey { case id, missionID = "mission_id", missionRunID = "mission_run_id", path, title, goal, state, since, holderDownSince = "holder_down_since" } } -public struct WorkResource: Codable, Sendable { public let missionID: String?; public let id, kind, revision, updatedAt, missionRunID, generationID, definitionID, path, state: String; public let gateKind, title, assignedTo, lastProgress, progressAt: String?; public let attempt: UInt; public let readinessEpoch: UInt64; public let claimant, claimIncarnation, submittedIncarnation, blockedReason: String?; public let agentless: Bool?; public let claimExpiresAtUnixMS, executionStartedAtUnixMS, nudgedAtUnixMS, executionElapsedMS, executionWaitedMS, timeoutMS: UInt64?; public let blockers, goals, constraints: [String]; public let personAnswers: [PersonAnswerRecord]?; public let usage: UsageSummary?; public let operational: Operational?; enum CodingKeys: String, CodingKey { case id, kind, revision, updatedAt = "updated_at", missionID = "mission_id", missionRunID = "mission_run_id", generationID = "generation_id", definitionID = "definition_id", path, state, gateKind = "gate_kind", title, assignedTo = "assigned_to", lastProgress = "last_progress", progressAt = "progress_at", attempt, readinessEpoch = "readiness_epoch", claimant, claimIncarnation = "claim_incarnation", submittedIncarnation = "submitted_incarnation", agentless, claimExpiresAtUnixMS = "claim_expires_at_unix_ms", executionStartedAtUnixMS = "execution_started_at_unix_ms", nudgedAtUnixMS = "nudged_at_unix_ms", executionElapsedMS = "execution_elapsed_ms", executionWaitedMS = "execution_waited_ms", timeoutMS = "timeout_ms", blockedReason = "blocked_reason", blockers, goals, constraints, personAnswers = "person_answers", usage, operational } } +public struct WorkResource: Codable, Sendable { public let missionID: String?; public let id, kind, revision, updatedAt, missionRunID, generationID, definitionID, path, state: String; public let gateKind, title, assignedTo, lastProgress, progressAt: String?; public let program: WorkProgram?; public let attempt: UInt; public let readinessEpoch: UInt64; public let claimant, claimIncarnation, submittedIncarnation, blockedReason: String?; public let agentless: Bool?; public let claimExpiresAtUnixMS, executionStartedAtUnixMS, nudgedAtUnixMS, executionElapsedMS, executionWaitedMS, timeoutMS: UInt64?; public let blockers, goals, constraints: [String]; public let personAnswers: [PersonAnswerRecord]?; public let usage: UsageSummary?; public let operational: Operational?; enum CodingKeys: String, CodingKey { case id, kind, revision, updatedAt = "updated_at", missionID = "mission_id", missionRunID = "mission_run_id", generationID = "generation_id", definitionID = "definition_id", path, state, gateKind = "gate_kind", program, title, assignedTo = "assigned_to", lastProgress = "last_progress", progressAt = "progress_at", attempt, readinessEpoch = "readiness_epoch", claimant, claimIncarnation = "claim_incarnation", submittedIncarnation = "submitted_incarnation", agentless, claimExpiresAtUnixMS = "claim_expires_at_unix_ms", executionStartedAtUnixMS = "execution_started_at_unix_ms", nudgedAtUnixMS = "nudged_at_unix_ms", executionElapsedMS = "execution_elapsed_ms", executionWaitedMS = "execution_waited_ms", timeoutMS = "timeout_ms", blockedReason = "blocked_reason", blockers, goals, constraints, personAnswers = "person_answers", usage, operational } } +/// A program step's innermost open journal key and journal timing. +public struct WorkProgram: Codable, Sendable { public let openKey, child: String?; public let openSinceUnixMS, lastEntryAtUnixMS: UInt64?; public let stalledAfterMS: UInt64; enum CodingKeys: String, CodingKey { case openKey = "open_key", child, openSinceUnixMS = "open_since_unix_ms", lastEntryAtUnixMS = "last_entry_at_unix_ms", stalledAfterMS = "stalled_after_ms" } } public struct AgentRelationship: Codable, Sendable { public let agentID: String; public let reason: String?; enum CodingKeys: String, CodingKey { case agentID = "agent_id", reason } } public enum ResourceReferenceKind: String, Codable, Sendable { case uri, subject } public enum ResourceReferenceResolution: String, Codable, Sendable { case resolved, unresolved; case notApplicable = "not-applicable" } diff --git a/clients/typescript/fractal-client/Models.generated.ts b/clients/typescript/fractal-client/Models.generated.ts index 19a491ae870..36a108f28ca 100644 --- a/clients/typescript/fractal-client/Models.generated.ts +++ b/clients/typescript/fractal-client/Models.generated.ts @@ -4508,6 +4508,7 @@ export type Work = ResourceHeader & { | "human" | "mixed" | "run" + | "program" | null ; generation_id: Id; @@ -4519,6 +4520,7 @@ export type Work = ResourceHeader & { nudged_at_unix_ms?: number | null; path: string; person_answers?: Array; + program?: (WorkProgram | null); progress_at?: (Timestamp | null); readiness_epoch: number; state: WorkState; @@ -4540,6 +4542,14 @@ export type WorkLabel = { title?: string | null; }; +export type WorkProgram = { + child?: string | null; + last_entry_at_unix_ms?: number | null; + open_key?: string | null; + open_since_unix_ms?: number | null; + stalled_after_ms: number; +}; + export type WorkState = "waiting-person" | "waiting" diff --git a/clients/typescript/fractal-client/ReadSchema.generated.ts b/clients/typescript/fractal-client/ReadSchema.generated.ts index b6b6a73dc95..c11b2a75b84 100644 --- a/clients/typescript/fractal-client/ReadSchema.generated.ts +++ b/clients/typescript/fractal-client/ReadSchema.generated.ts @@ -2983,6 +2983,19 @@ export const PersonAnswerRecord = /*#__PURE__*/ (() => Schema.Struct({ export type PersonAnswerRecord = typeof PersonAnswerRecord.Type export type PersonAnswerRecordEncoded = typeof PersonAnswerRecord.Encoded +export const WorkProgram = /*#__PURE__*/ (() => Schema.Struct({ + /** The child mission run or ask subject the open key waits on. */ + "child": Schema.OptionFromOptionalNullOr(Schema.String, NULL_NONE).annotate({ description: "The child mission run or ask subject the open key waits on." }), + "last_entry_at_unix_ms": Schema.OptionFromOptionalNullOr(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)).pipe(Schema.decodeTo(Schema.DateTimeUtcFromMillis.check(epochRange))), NULL_NONE), + /** The most recently started journal key without a result: the innermost open key. */ + "open_key": Schema.OptionFromOptionalNullOr(Schema.String, NULL_NONE).annotate({ description: "The most recently started journal key without a result: the innermost open key." }), + "open_since_unix_ms": Schema.OptionFromOptionalNullOr(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)).pipe(Schema.decodeTo(Schema.DateTimeUtcFromMillis.check(epochRange))), NULL_NONE), + /** How long the journal may go without a new entry before the step has made no progress. */ + "stalled_after_ms": Schema.Int.check(Schema.isGreaterThanOrEqualTo(1)).pipe(Schema.decodeTo(Schema.Duration.check(durationMillisRange), wholeUnits(1))).annotate({ description: "How long the journal may go without a new entry before the step has made no progress." }) +}).annotate({ identifier: "WorkProgram" }))() +export type WorkProgram = typeof WorkProgram.Type +export type WorkProgramEncoded = typeof WorkProgram.Encoded + export const Work = /*#__PURE__*/ (() => Schema.Struct({ "agentless": optionalKey(Schema.Boolean), "assigned_to": Schema.OptionFromOptionalNullOr(Id, NULL_NONE), @@ -2998,7 +3011,16 @@ export const Work = /*#__PURE__*/ (() => Schema.Struct({ "execution_started_at_unix_ms": Schema.OptionFromOptionalNullOr(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)).pipe(Schema.decodeTo(Schema.DateTimeUtcFromMillis.check(epochRange))), NULL_NONE), /** Accumulated execution wait against this attempt's wait budget. */ "execution_waited_ms": optionalKey(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)).pipe(Schema.decodeTo(Schema.Duration.check(durationMillisRange), wholeUnits(1)))).annotate({ description: "Accumulated execution wait against this attempt's wait budget." }), - "gate_kind": Schema.OptionFromOptionalNullOr(Schema.Literals(["watch","predicate","command","llm","human","mixed","run"]), NULL_NONE), + "gate_kind": Schema.OptionFromOptionalNullOr(Schema.Literals([ + "watch", + "predicate", + "command", + "llm", + "human", + "mixed", + "run", + "program" +]), NULL_NONE), "generation_id": Id, "goals": Schema.Array(Schema.String), "id": Id, @@ -3013,6 +3035,8 @@ export const Work = /*#__PURE__*/ (() => Schema.Struct({ "path": Schema.String, /** Responses to the asks this attempt made, or a person ask's own response. */ "person_answers": optionalKey(Schema.Array(PersonAnswerRecord)).annotate({ description: "Responses to the asks this attempt made, or a person ask's own response." }), + /** A program step's innermost open journal key and journal timing; null for other work. */ + "program": Schema.OptionFromOptionalNullOr(WorkProgram, NULL_NONE).annotate({ description: "A program step's innermost open journal key and journal timing; null for other work." }), /** Current attempt latest nonempty work.progress summary time; not lease renewal or seat activity. Lightweight mission list cards are unhydrated and return null, not proof of no progress. */ "progress_at": Schema.OptionFromOptionalNullOr(Timestamp, NULL_NONE).annotate({ description: "Current attempt latest nonempty work.progress summary time; not lease renewal or seat activity. Lightweight mission list cards are unhydrated and return null, not proof of no progress." }), "readiness_epoch": Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)), diff --git a/clients/typescript/fractal-client/Schema.generated.ts b/clients/typescript/fractal-client/Schema.generated.ts index bcae26143dc..c7216188d5b 100644 --- a/clients/typescript/fractal-client/Schema.generated.ts +++ b/clients/typescript/fractal-client/Schema.generated.ts @@ -10,8 +10,8 @@ import type { SubjectRef, UnknownCase } from "./SchemaSupport.generated.ts" export { API_VERSION, containsUnknownCase, decodeUnknownSync, decodeUnknownEffect } from "./SchemaSupport.generated.ts" export type { SubjectRef, UnknownCase, DecodeMode } from "./SchemaSupport.generated.ts" -import { ActorRef, Agent, AgentCheckout, AgentDeclaration, AgentDelivery, AgentEndReason, AgentHandoff, AgentId, AgentLifecycle, AgentMailCustody, AgentMessageDay, AgentMessageEstimate, AgentPresentation, AgentPresentationNow, AgentPresentationWork, AgentQueue, AgentQueueMove, AgentQueueRun, AgentReachability, AgentRelatedRun, AgentRelatedRuns, AgentRelationship, AgentRepository, AgentState, AgentStopping, AgentSubagent, AgentSuspension, AgentTodo, AgentWorkspace, AgentsChanges, AgentsPublicationMetadata, AgentsPublicationSubscribe, AgentsSnapshotChunk, AgentsStatusWatermark, Arrangement, ArrangementBody, ArrangementBodyV1, ArrangementBodyV2, ArrangementFolder, ArrangementFolderId, ArrangementId, ArrangementKey, ArrangementName, ArrangementNameRegister, ArrangementPage, ArrangementPerson, ArrangementPlacement, ArrangementPlacementRegister, ArrangementPosition, ArrangementPositionRegister, ArrangementResolved, ArrangementSubject, ArrangementTombstoneRegister, AskDefaultAnswer, AssistantAgentDetailRequest, AssistantAgentItem, AssistantAgentPage, AssistantAttentionItem, AssistantAttentionPage, AssistantContentRange, AssistantConversationContent, AssistantConversationContentRequest, AssistantConversationSearchRequest, AssistantConversationTimelineRequest, AssistantMinimalReadMode, AssistantMissionItem, AssistantMissionPage, AssistantPageRequest, AssistantPrincipal, AssistantProjection, AssistantResponseBytes, AssistantSenderClass, AssistantSummary, AssistantSummaryRequest, Attachment, Attention, AttentionActor, AttentionBlocked, AttentionBudgetDeclared, AttentionBudgetReport, AttentionBudgetRequestRef, AttentionLaunchAgent, AttentionLaunchAssignee, AttentionLaunchPreview, AttentionLaunchStep, AttentionLaunchUnavailable, AttentionLaunchUnavailableReason, AttentionPresentation, AttentionPrompt, AttentionPromptCall, AttentionRevision, AttentionRevisionLine, AttentionTargetState, AttentionWaiting, BlobChunk, CanonicalNode, Capabilities, Capability, ClientConnection, ClientConnections, CollectionName, ConnectedClient, ConversationChanges, ConversationContentChunk, ConversationContentRef, ConversationHeader, ConversationHeaderField, ConversationSearch, ConversationSearchHit, Cursor, DecisionOption, DecisionResponse, DeclaredAdmission, DeclaredAttentionBudget, DeclaredBudgets, DeclaredLane, DeclaredPriority, DeclaredResource, Device, DeviceCommand, DeviceCommandState, Diagnostic, DirectiveNote, DirectiveNotes, DocumentContent, DocumentItem, DocumentReference, DocumentReferences, DocumentsPage, ErrorCode, EventPage, ExchangeTransport, Fence, Glass, GlassBody, GlassLayout, GlassTab, HarnessInventory, HarnessInventoryItem, HarnessPhase, HarnessQuestion, HarnessQuestionAnswerability, HarnessQuestionAnswerabilityReason, HarnessQuestionItem, HarnessQuestionResolution, HarnessResolvedAnswer, HarnessTask, HarnessTodoSnapshot, HarnessTodoTotals, History, HostId, HostRepositories, Id, JobSummary, Lane, LaneChange, LaneEntry, Launch, LaunchApproval, LaunchDecision, LaunchPreview, LaunchVariant, Machine, MachineCapacity, MachineOccupancy, MachineTransport, MailBacklog, Message, MessageBody, MessageProvenance, MessageStatusFrom, Mission, MissionActionOffer, MissionDeclaration, MissionDeclarationProvenance, MissionId, MissionPresentation, MissionQueuedBehind, MissionRunId, MissionRunOutcome, MissionRunSummary, MissionState, MissionStep, MissionStepAttempt, MissionStepEta, MissionStepPresentation, MissionStepTransition, MissionStepWait, MissionStuck, MissionWake, MustAct, Observer, ObserverSpec, Operation, Operational, OrderedMembership, OrderedMembershipPage, OrderedMembershipPosition, OrderedMembershipState, OwnedSet, Page, PageInfo, PairingAdvertisement, ParticipantRef, PersonAnswer, PersonAnswerRecord, PersonId, PersonOptionNote, PersonUpdate, PhoneInspectArgs, PhoneMessageReadArgs, PhoneNavigateArgs, PhoneRouteString, PlannerConfig, PresentationActor, PrivateNotesAction, PrivateNotesData, PrivateNotesFence, PrivateNotesSubject, PrivateNotesUri, ProjectionEvent, PublicationDefinition, PushEnvironment, PushRegistrationStatus, ReadCollectionCommand, ReadCollectionFrame, RelatedRunWait, ReplicatedNotice, RequestAnswerOption, RequestSubject, Resource, ResourceCollectionName, ResourceHeader, ResourceObservation, ResourcesPage, RespondentRef, Revision, RunGenerationId, RunOutcomeStatus, Runtime, RuntimeId, ScgReadAuthorization, ScgReadAuthorizationPayload, ScheduleCoverage, ScheduleDeclaration, ScheduleKnowledge, ScheduleOwner, ScheduleUnheardHost, SchedulesPage, SenderRef, Session, SloHistory, SloList, SloRow, Snapshot, SnapshotId, StatusHistory, StatusTransition, StepRunId, StreamSync, StructuredDiff, StructuredRequest, SubagentSummary, SubjectDefinition, SubscribeTraceContext, Subscription, SubscriptionSpec, Summary, SummaryMachines, SyncNotice, SyncPeer, TimelineAskAnswer, TimelineAskQuestion, TimelineBlock, TimelineContentBody, TimelineDelta, TimelineEntry, TimelineEntryId, TimelineErrorBody, TimelineMessageBody, TimelinePage, TimelineRecipientRef, TimelineRedactionBody, TimelineStatus, TimelineStatusBody, TimelineTodoItem, TimelineToolCallBody, TimelineToolResultBody, TimelineTruncationBody, TimelineUsageBody, TimelineView, TimelineViewAsk, TimelineViewAssistantError, TimelineViewBash, TimelineViewCompaction, TimelineViewCredentialPin, TimelineViewEdit, TimelineViewEval, TimelineViewGeneric, TimelineViewHub, TimelineViewIrc, TimelineViewJob, TimelineViewModelChange, TimelineViewOmpEvent, TimelineViewRead, TimelineViewResetBoundary, TimelineViewSearch, TimelineViewSessionExit, TimelineViewSkill, TimelineViewTask, TimelineViewThinkingLevel, TimelineViewTitle, TimelineViewTodo, TimelineViewToolStart, TimelineViewWrite, Timestamp, UnknownResource, Usage, UsageAccount, UsageContext, UsageCredits, UsageLimit, UsagePeriod, UsagePricing, UsagePricingRates, UsageRow, UsageSummary, Value, Visualization, VisualizationEdge, VisualizationGroup, VisualizationNode, Work, WorkLabel, WorkState } from "./ReadSchema.generated.ts" -import type { ActorRefEncoded, AgentEncoded, AgentCheckoutEncoded, AgentDeclarationEncoded, AgentDeliveryEncoded, AgentEndReasonEncoded, AgentHandoffEncoded, AgentIdEncoded, AgentLifecycleEncoded, AgentMailCustodyEncoded, AgentMessageDayEncoded, AgentMessageEstimateEncoded, AgentPresentationEncoded, AgentPresentationNowEncoded, AgentPresentationWorkEncoded, AgentQueueEncoded, AgentQueueMoveEncoded, AgentQueueRunEncoded, AgentReachabilityEncoded, AgentRelatedRunEncoded, AgentRelatedRunsEncoded, AgentRelationshipEncoded, AgentRepositoryEncoded, AgentStateEncoded, AgentStoppingEncoded, AgentSubagentEncoded, AgentSuspensionEncoded, AgentTodoEncoded, AgentWorkspaceEncoded, AgentsChangesEncoded, AgentsPublicationMetadataEncoded, AgentsPublicationSubscribeEncoded, AgentsSnapshotChunkEncoded, AgentsStatusWatermarkEncoded, ArrangementEncoded, ArrangementBodyEncoded, ArrangementBodyV1Encoded, ArrangementBodyV2Encoded, ArrangementFolderEncoded, ArrangementFolderIdEncoded, ArrangementIdEncoded, ArrangementKeyEncoded, ArrangementNameEncoded, ArrangementNameRegisterEncoded, ArrangementPageEncoded, ArrangementPersonEncoded, ArrangementPlacementEncoded, ArrangementPlacementRegisterEncoded, ArrangementPositionEncoded, ArrangementPositionRegisterEncoded, ArrangementResolvedEncoded, ArrangementSubjectEncoded, ArrangementTombstoneRegisterEncoded, AskDefaultAnswerEncoded, AssistantAgentDetailRequestEncoded, AssistantAgentItemEncoded, AssistantAgentPageEncoded, AssistantAttentionItemEncoded, AssistantAttentionPageEncoded, AssistantContentRangeEncoded, AssistantConversationContentEncoded, AssistantConversationContentRequestEncoded, AssistantConversationSearchRequestEncoded, AssistantConversationTimelineRequestEncoded, AssistantMinimalReadModeEncoded, AssistantMissionItemEncoded, AssistantMissionPageEncoded, AssistantPageRequestEncoded, AssistantPrincipalEncoded, AssistantProjectionEncoded, AssistantResponseBytesEncoded, AssistantSenderClassEncoded, AssistantSummaryEncoded, AssistantSummaryRequestEncoded, AttachmentEncoded, AttentionEncoded, AttentionActorEncoded, AttentionBlockedEncoded, AttentionBudgetDeclaredEncoded, AttentionBudgetReportEncoded, AttentionBudgetRequestRefEncoded, AttentionLaunchAgentEncoded, AttentionLaunchAssigneeEncoded, AttentionLaunchPreviewEncoded, AttentionLaunchStepEncoded, AttentionLaunchUnavailableEncoded, AttentionLaunchUnavailableReasonEncoded, AttentionPresentationEncoded, AttentionPromptEncoded, AttentionPromptCallEncoded, AttentionRevisionEncoded, AttentionRevisionLineEncoded, AttentionTargetStateEncoded, AttentionWaitingEncoded, BlobChunkEncoded, CanonicalNodeEncoded, CapabilitiesEncoded, CapabilityEncoded, ClientConnectionEncoded, ClientConnectionsEncoded, CollectionNameEncoded, ConnectedClientEncoded, ConversationChangesEncoded, ConversationContentChunkEncoded, ConversationContentRefEncoded, ConversationHeaderEncoded, ConversationHeaderFieldEncoded, ConversationSearchEncoded, ConversationSearchHitEncoded, CursorEncoded, DecisionOptionEncoded, DecisionResponseEncoded, DeclaredAdmissionEncoded, DeclaredAttentionBudgetEncoded, DeclaredBudgetsEncoded, DeclaredLaneEncoded, DeclaredPriorityEncoded, DeclaredResourceEncoded, DeviceEncoded, DeviceCommandEncoded, DeviceCommandStateEncoded, DiagnosticEncoded, DirectiveNoteEncoded, DirectiveNotesEncoded, DocumentContentEncoded, DocumentItemEncoded, DocumentReferenceEncoded, DocumentReferencesEncoded, DocumentsPageEncoded, ErrorCodeEncoded, EventPageEncoded, ExchangeTransportEncoded, FenceEncoded, GlassEncoded, GlassBodyEncoded, GlassLayoutEncoded, GlassTabEncoded, HarnessInventoryEncoded, HarnessInventoryItemEncoded, HarnessPhaseEncoded, HarnessQuestionEncoded, HarnessQuestionAnswerabilityEncoded, HarnessQuestionAnswerabilityReasonEncoded, HarnessQuestionItemEncoded, HarnessQuestionResolutionEncoded, HarnessResolvedAnswerEncoded, HarnessTaskEncoded, HarnessTodoSnapshotEncoded, HarnessTodoTotalsEncoded, HistoryEncoded, HostIdEncoded, HostRepositoriesEncoded, IdEncoded, JobSummaryEncoded, LaneEncoded, LaneChangeEncoded, LaneEntryEncoded, LaunchEncoded, LaunchApprovalEncoded, LaunchDecisionEncoded, LaunchPreviewEncoded, LaunchVariantEncoded, MachineEncoded, MachineCapacityEncoded, MachineOccupancyEncoded, MachineTransportEncoded, MailBacklogEncoded, MessageEncoded, MessageBodyEncoded, MessageProvenanceEncoded, MessageStatusFromEncoded, MissionEncoded, MissionActionOfferEncoded, MissionDeclarationEncoded, MissionDeclarationProvenanceEncoded, MissionIdEncoded, MissionPresentationEncoded, MissionQueuedBehindEncoded, MissionRunIdEncoded, MissionRunOutcomeEncoded, MissionRunSummaryEncoded, MissionStateEncoded, MissionStepEncoded, MissionStepAttemptEncoded, MissionStepEtaEncoded, MissionStepPresentationEncoded, MissionStepTransitionEncoded, MissionStepWaitEncoded, MissionStuckEncoded, MissionWakeEncoded, MustActEncoded, ObserverEncoded, ObserverSpecEncoded, OperationEncoded, OperationalEncoded, OrderedMembershipEncoded, OrderedMembershipPageEncoded, OrderedMembershipPositionEncoded, OrderedMembershipStateEncoded, OwnedSetEncoded, PageEncoded, PageInfoEncoded, PairingAdvertisementEncoded, ParticipantRefEncoded, PersonAnswerEncoded, PersonAnswerRecordEncoded, PersonIdEncoded, PersonOptionNoteEncoded, PersonUpdateEncoded, PhoneInspectArgsEncoded, PhoneMessageReadArgsEncoded, PhoneNavigateArgsEncoded, PhoneRouteStringEncoded, PlannerConfigEncoded, PresentationActorEncoded, PrivateNotesActionEncoded, PrivateNotesDataEncoded, PrivateNotesFenceEncoded, PrivateNotesSubjectEncoded, PrivateNotesUriEncoded, ProjectionEventEncoded, PublicationDefinitionEncoded, PushEnvironmentEncoded, PushRegistrationStatusEncoded, ReadCollectionCommandEncoded, ReadCollectionFrameEncoded, RelatedRunWaitEncoded, ReplicatedNoticeEncoded, RequestAnswerOptionEncoded, RequestSubjectEncoded, ResourceEncoded, ResourceCollectionNameEncoded, ResourceHeaderEncoded, ResourceObservationEncoded, ResourcesPageEncoded, RespondentRefEncoded, RevisionEncoded, RunGenerationIdEncoded, RunOutcomeStatusEncoded, RuntimeEncoded, RuntimeIdEncoded, ScgReadAuthorizationEncoded, ScgReadAuthorizationPayloadEncoded, ScheduleCoverageEncoded, ScheduleDeclarationEncoded, ScheduleKnowledgeEncoded, ScheduleOwnerEncoded, ScheduleUnheardHostEncoded, SchedulesPageEncoded, SenderRefEncoded, SessionEncoded, SloHistoryEncoded, SloListEncoded, SloRowEncoded, SnapshotEncoded, SnapshotIdEncoded, StatusHistoryEncoded, StatusTransitionEncoded, StepRunIdEncoded, StreamSyncEncoded, StructuredDiffEncoded, StructuredRequestEncoded, SubagentSummaryEncoded, SubjectDefinitionEncoded, SubscribeTraceContextEncoded, SubscriptionEncoded, SubscriptionSpecEncoded, SummaryEncoded, SummaryMachinesEncoded, SyncNoticeEncoded, SyncPeerEncoded, TimelineAskAnswerEncoded, TimelineAskQuestionEncoded, TimelineBlockEncoded, TimelineContentBodyEncoded, TimelineDeltaEncoded, TimelineEntryEncoded, TimelineEntryIdEncoded, TimelineErrorBodyEncoded, TimelineMessageBodyEncoded, TimelinePageEncoded, TimelineRecipientRefEncoded, TimelineRedactionBodyEncoded, TimelineStatusEncoded, TimelineStatusBodyEncoded, TimelineTodoItemEncoded, TimelineToolCallBodyEncoded, TimelineToolResultBodyEncoded, TimelineTruncationBodyEncoded, TimelineUsageBodyEncoded, TimelineViewEncoded, TimelineViewAskEncoded, TimelineViewAssistantErrorEncoded, TimelineViewBashEncoded, TimelineViewCompactionEncoded, TimelineViewCredentialPinEncoded, TimelineViewEditEncoded, TimelineViewEvalEncoded, TimelineViewGenericEncoded, TimelineViewHubEncoded, TimelineViewIrcEncoded, TimelineViewJobEncoded, TimelineViewModelChangeEncoded, TimelineViewOmpEventEncoded, TimelineViewReadEncoded, TimelineViewResetBoundaryEncoded, TimelineViewSearchEncoded, TimelineViewSessionExitEncoded, TimelineViewSkillEncoded, TimelineViewTaskEncoded, TimelineViewThinkingLevelEncoded, TimelineViewTitleEncoded, TimelineViewTodoEncoded, TimelineViewToolStartEncoded, TimelineViewWriteEncoded, TimestampEncoded, UnknownResourceEncoded, UsageEncoded, UsageAccountEncoded, UsageContextEncoded, UsageCreditsEncoded, UsageLimitEncoded, UsagePeriodEncoded, UsagePricingEncoded, UsagePricingRatesEncoded, UsageRowEncoded, UsageSummaryEncoded, ValueEncoded, VisualizationEncoded, VisualizationEdgeEncoded, VisualizationGroupEncoded, VisualizationNodeEncoded, WorkEncoded, WorkLabelEncoded, WorkStateEncoded } from "./ReadSchema.generated.ts" +import { ActorRef, Agent, AgentCheckout, AgentDeclaration, AgentDelivery, AgentEndReason, AgentHandoff, AgentId, AgentLifecycle, AgentMailCustody, AgentMessageDay, AgentMessageEstimate, AgentPresentation, AgentPresentationNow, AgentPresentationWork, AgentQueue, AgentQueueMove, AgentQueueRun, AgentReachability, AgentRelatedRun, AgentRelatedRuns, AgentRelationship, AgentRepository, AgentState, AgentStopping, AgentSubagent, AgentSuspension, AgentTodo, AgentWorkspace, AgentsChanges, AgentsPublicationMetadata, AgentsPublicationSubscribe, AgentsSnapshotChunk, AgentsStatusWatermark, Arrangement, ArrangementBody, ArrangementBodyV1, ArrangementBodyV2, ArrangementFolder, ArrangementFolderId, ArrangementId, ArrangementKey, ArrangementName, ArrangementNameRegister, ArrangementPage, ArrangementPerson, ArrangementPlacement, ArrangementPlacementRegister, ArrangementPosition, ArrangementPositionRegister, ArrangementResolved, ArrangementSubject, ArrangementTombstoneRegister, AskDefaultAnswer, AssistantAgentDetailRequest, AssistantAgentItem, AssistantAgentPage, AssistantAttentionItem, AssistantAttentionPage, AssistantContentRange, AssistantConversationContent, AssistantConversationContentRequest, AssistantConversationSearchRequest, AssistantConversationTimelineRequest, AssistantMinimalReadMode, AssistantMissionItem, AssistantMissionPage, AssistantPageRequest, AssistantPrincipal, AssistantProjection, AssistantResponseBytes, AssistantSenderClass, AssistantSummary, AssistantSummaryRequest, Attachment, Attention, AttentionActor, AttentionBlocked, AttentionBudgetDeclared, AttentionBudgetReport, AttentionBudgetRequestRef, AttentionLaunchAgent, AttentionLaunchAssignee, AttentionLaunchPreview, AttentionLaunchStep, AttentionLaunchUnavailable, AttentionLaunchUnavailableReason, AttentionPresentation, AttentionPrompt, AttentionPromptCall, AttentionRevision, AttentionRevisionLine, AttentionTargetState, AttentionWaiting, BlobChunk, CanonicalNode, Capabilities, Capability, ClientConnection, ClientConnections, CollectionName, ConnectedClient, ConversationChanges, ConversationContentChunk, ConversationContentRef, ConversationHeader, ConversationHeaderField, ConversationSearch, ConversationSearchHit, Cursor, DecisionOption, DecisionResponse, DeclaredAdmission, DeclaredAttentionBudget, DeclaredBudgets, DeclaredLane, DeclaredPriority, DeclaredResource, Device, DeviceCommand, DeviceCommandState, Diagnostic, DirectiveNote, DirectiveNotes, DocumentContent, DocumentItem, DocumentReference, DocumentReferences, DocumentsPage, ErrorCode, EventPage, ExchangeTransport, Fence, Glass, GlassBody, GlassLayout, GlassTab, HarnessInventory, HarnessInventoryItem, HarnessPhase, HarnessQuestion, HarnessQuestionAnswerability, HarnessQuestionAnswerabilityReason, HarnessQuestionItem, HarnessQuestionResolution, HarnessResolvedAnswer, HarnessTask, HarnessTodoSnapshot, HarnessTodoTotals, History, HostId, HostRepositories, Id, JobSummary, Lane, LaneChange, LaneEntry, Launch, LaunchApproval, LaunchDecision, LaunchPreview, LaunchVariant, Machine, MachineCapacity, MachineOccupancy, MachineTransport, MailBacklog, Message, MessageBody, MessageProvenance, MessageStatusFrom, Mission, MissionActionOffer, MissionDeclaration, MissionDeclarationProvenance, MissionId, MissionPresentation, MissionQueuedBehind, MissionRunId, MissionRunOutcome, MissionRunSummary, MissionState, MissionStep, MissionStepAttempt, MissionStepEta, MissionStepPresentation, MissionStepTransition, MissionStepWait, MissionStuck, MissionWake, MustAct, Observer, ObserverSpec, Operation, Operational, OrderedMembership, OrderedMembershipPage, OrderedMembershipPosition, OrderedMembershipState, OwnedSet, Page, PageInfo, PairingAdvertisement, ParticipantRef, PersonAnswer, PersonAnswerRecord, PersonId, PersonOptionNote, PersonUpdate, PhoneInspectArgs, PhoneMessageReadArgs, PhoneNavigateArgs, PhoneRouteString, PlannerConfig, PresentationActor, PrivateNotesAction, PrivateNotesData, PrivateNotesFence, PrivateNotesSubject, PrivateNotesUri, ProjectionEvent, PublicationDefinition, PushEnvironment, PushRegistrationStatus, ReadCollectionCommand, ReadCollectionFrame, RelatedRunWait, ReplicatedNotice, RequestAnswerOption, RequestSubject, Resource, ResourceCollectionName, ResourceHeader, ResourceObservation, ResourcesPage, RespondentRef, Revision, RunGenerationId, RunOutcomeStatus, Runtime, RuntimeId, ScgReadAuthorization, ScgReadAuthorizationPayload, ScheduleCoverage, ScheduleDeclaration, ScheduleKnowledge, ScheduleOwner, ScheduleUnheardHost, SchedulesPage, SenderRef, Session, SloHistory, SloList, SloRow, Snapshot, SnapshotId, StatusHistory, StatusTransition, StepRunId, StreamSync, StructuredDiff, StructuredRequest, SubagentSummary, SubjectDefinition, SubscribeTraceContext, Subscription, SubscriptionSpec, Summary, SummaryMachines, SyncNotice, SyncPeer, TimelineAskAnswer, TimelineAskQuestion, TimelineBlock, TimelineContentBody, TimelineDelta, TimelineEntry, TimelineEntryId, TimelineErrorBody, TimelineMessageBody, TimelinePage, TimelineRecipientRef, TimelineRedactionBody, TimelineStatus, TimelineStatusBody, TimelineTodoItem, TimelineToolCallBody, TimelineToolResultBody, TimelineTruncationBody, TimelineUsageBody, TimelineView, TimelineViewAsk, TimelineViewAssistantError, TimelineViewBash, TimelineViewCompaction, TimelineViewCredentialPin, TimelineViewEdit, TimelineViewEval, TimelineViewGeneric, TimelineViewHub, TimelineViewIrc, TimelineViewJob, TimelineViewModelChange, TimelineViewOmpEvent, TimelineViewRead, TimelineViewResetBoundary, TimelineViewSearch, TimelineViewSessionExit, TimelineViewSkill, TimelineViewTask, TimelineViewThinkingLevel, TimelineViewTitle, TimelineViewTodo, TimelineViewToolStart, TimelineViewWrite, Timestamp, UnknownResource, Usage, UsageAccount, UsageContext, UsageCredits, UsageLimit, UsagePeriod, UsagePricing, UsagePricingRates, UsageRow, UsageSummary, Value, Visualization, VisualizationEdge, VisualizationGroup, VisualizationNode, Work, WorkLabel, WorkProgram, WorkState } from "./ReadSchema.generated.ts" +import type { ActorRefEncoded, AgentEncoded, AgentCheckoutEncoded, AgentDeclarationEncoded, AgentDeliveryEncoded, AgentEndReasonEncoded, AgentHandoffEncoded, AgentIdEncoded, AgentLifecycleEncoded, AgentMailCustodyEncoded, AgentMessageDayEncoded, AgentMessageEstimateEncoded, AgentPresentationEncoded, AgentPresentationNowEncoded, AgentPresentationWorkEncoded, AgentQueueEncoded, AgentQueueMoveEncoded, AgentQueueRunEncoded, AgentReachabilityEncoded, AgentRelatedRunEncoded, AgentRelatedRunsEncoded, AgentRelationshipEncoded, AgentRepositoryEncoded, AgentStateEncoded, AgentStoppingEncoded, AgentSubagentEncoded, AgentSuspensionEncoded, AgentTodoEncoded, AgentWorkspaceEncoded, AgentsChangesEncoded, AgentsPublicationMetadataEncoded, AgentsPublicationSubscribeEncoded, AgentsSnapshotChunkEncoded, AgentsStatusWatermarkEncoded, ArrangementEncoded, ArrangementBodyEncoded, ArrangementBodyV1Encoded, ArrangementBodyV2Encoded, ArrangementFolderEncoded, ArrangementFolderIdEncoded, ArrangementIdEncoded, ArrangementKeyEncoded, ArrangementNameEncoded, ArrangementNameRegisterEncoded, ArrangementPageEncoded, ArrangementPersonEncoded, ArrangementPlacementEncoded, ArrangementPlacementRegisterEncoded, ArrangementPositionEncoded, ArrangementPositionRegisterEncoded, ArrangementResolvedEncoded, ArrangementSubjectEncoded, ArrangementTombstoneRegisterEncoded, AskDefaultAnswerEncoded, AssistantAgentDetailRequestEncoded, AssistantAgentItemEncoded, AssistantAgentPageEncoded, AssistantAttentionItemEncoded, AssistantAttentionPageEncoded, AssistantContentRangeEncoded, AssistantConversationContentEncoded, AssistantConversationContentRequestEncoded, AssistantConversationSearchRequestEncoded, AssistantConversationTimelineRequestEncoded, AssistantMinimalReadModeEncoded, AssistantMissionItemEncoded, AssistantMissionPageEncoded, AssistantPageRequestEncoded, AssistantPrincipalEncoded, AssistantProjectionEncoded, AssistantResponseBytesEncoded, AssistantSenderClassEncoded, AssistantSummaryEncoded, AssistantSummaryRequestEncoded, AttachmentEncoded, AttentionEncoded, AttentionActorEncoded, AttentionBlockedEncoded, AttentionBudgetDeclaredEncoded, AttentionBudgetReportEncoded, AttentionBudgetRequestRefEncoded, AttentionLaunchAgentEncoded, AttentionLaunchAssigneeEncoded, AttentionLaunchPreviewEncoded, AttentionLaunchStepEncoded, AttentionLaunchUnavailableEncoded, AttentionLaunchUnavailableReasonEncoded, AttentionPresentationEncoded, AttentionPromptEncoded, AttentionPromptCallEncoded, AttentionRevisionEncoded, AttentionRevisionLineEncoded, AttentionTargetStateEncoded, AttentionWaitingEncoded, BlobChunkEncoded, CanonicalNodeEncoded, CapabilitiesEncoded, CapabilityEncoded, ClientConnectionEncoded, ClientConnectionsEncoded, CollectionNameEncoded, ConnectedClientEncoded, ConversationChangesEncoded, ConversationContentChunkEncoded, ConversationContentRefEncoded, ConversationHeaderEncoded, ConversationHeaderFieldEncoded, ConversationSearchEncoded, ConversationSearchHitEncoded, CursorEncoded, DecisionOptionEncoded, DecisionResponseEncoded, DeclaredAdmissionEncoded, DeclaredAttentionBudgetEncoded, DeclaredBudgetsEncoded, DeclaredLaneEncoded, DeclaredPriorityEncoded, DeclaredResourceEncoded, DeviceEncoded, DeviceCommandEncoded, DeviceCommandStateEncoded, DiagnosticEncoded, DirectiveNoteEncoded, DirectiveNotesEncoded, DocumentContentEncoded, DocumentItemEncoded, DocumentReferenceEncoded, DocumentReferencesEncoded, DocumentsPageEncoded, ErrorCodeEncoded, EventPageEncoded, ExchangeTransportEncoded, FenceEncoded, GlassEncoded, GlassBodyEncoded, GlassLayoutEncoded, GlassTabEncoded, HarnessInventoryEncoded, HarnessInventoryItemEncoded, HarnessPhaseEncoded, HarnessQuestionEncoded, HarnessQuestionAnswerabilityEncoded, HarnessQuestionAnswerabilityReasonEncoded, HarnessQuestionItemEncoded, HarnessQuestionResolutionEncoded, HarnessResolvedAnswerEncoded, HarnessTaskEncoded, HarnessTodoSnapshotEncoded, HarnessTodoTotalsEncoded, HistoryEncoded, HostIdEncoded, HostRepositoriesEncoded, IdEncoded, JobSummaryEncoded, LaneEncoded, LaneChangeEncoded, LaneEntryEncoded, LaunchEncoded, LaunchApprovalEncoded, LaunchDecisionEncoded, LaunchPreviewEncoded, LaunchVariantEncoded, MachineEncoded, MachineCapacityEncoded, MachineOccupancyEncoded, MachineTransportEncoded, MailBacklogEncoded, MessageEncoded, MessageBodyEncoded, MessageProvenanceEncoded, MessageStatusFromEncoded, MissionEncoded, MissionActionOfferEncoded, MissionDeclarationEncoded, MissionDeclarationProvenanceEncoded, MissionIdEncoded, MissionPresentationEncoded, MissionQueuedBehindEncoded, MissionRunIdEncoded, MissionRunOutcomeEncoded, MissionRunSummaryEncoded, MissionStateEncoded, MissionStepEncoded, MissionStepAttemptEncoded, MissionStepEtaEncoded, MissionStepPresentationEncoded, MissionStepTransitionEncoded, MissionStepWaitEncoded, MissionStuckEncoded, MissionWakeEncoded, MustActEncoded, ObserverEncoded, ObserverSpecEncoded, OperationEncoded, OperationalEncoded, OrderedMembershipEncoded, OrderedMembershipPageEncoded, OrderedMembershipPositionEncoded, OrderedMembershipStateEncoded, OwnedSetEncoded, PageEncoded, PageInfoEncoded, PairingAdvertisementEncoded, ParticipantRefEncoded, PersonAnswerEncoded, PersonAnswerRecordEncoded, PersonIdEncoded, PersonOptionNoteEncoded, PersonUpdateEncoded, PhoneInspectArgsEncoded, PhoneMessageReadArgsEncoded, PhoneNavigateArgsEncoded, PhoneRouteStringEncoded, PlannerConfigEncoded, PresentationActorEncoded, PrivateNotesActionEncoded, PrivateNotesDataEncoded, PrivateNotesFenceEncoded, PrivateNotesSubjectEncoded, PrivateNotesUriEncoded, ProjectionEventEncoded, PublicationDefinitionEncoded, PushEnvironmentEncoded, PushRegistrationStatusEncoded, ReadCollectionCommandEncoded, ReadCollectionFrameEncoded, RelatedRunWaitEncoded, ReplicatedNoticeEncoded, RequestAnswerOptionEncoded, RequestSubjectEncoded, ResourceEncoded, ResourceCollectionNameEncoded, ResourceHeaderEncoded, ResourceObservationEncoded, ResourcesPageEncoded, RespondentRefEncoded, RevisionEncoded, RunGenerationIdEncoded, RunOutcomeStatusEncoded, RuntimeEncoded, RuntimeIdEncoded, ScgReadAuthorizationEncoded, ScgReadAuthorizationPayloadEncoded, ScheduleCoverageEncoded, ScheduleDeclarationEncoded, ScheduleKnowledgeEncoded, ScheduleOwnerEncoded, ScheduleUnheardHostEncoded, SchedulesPageEncoded, SenderRefEncoded, SessionEncoded, SloHistoryEncoded, SloListEncoded, SloRowEncoded, SnapshotEncoded, SnapshotIdEncoded, StatusHistoryEncoded, StatusTransitionEncoded, StepRunIdEncoded, StreamSyncEncoded, StructuredDiffEncoded, StructuredRequestEncoded, SubagentSummaryEncoded, SubjectDefinitionEncoded, SubscribeTraceContextEncoded, SubscriptionEncoded, SubscriptionSpecEncoded, SummaryEncoded, SummaryMachinesEncoded, SyncNoticeEncoded, SyncPeerEncoded, TimelineAskAnswerEncoded, TimelineAskQuestionEncoded, TimelineBlockEncoded, TimelineContentBodyEncoded, TimelineDeltaEncoded, TimelineEntryEncoded, TimelineEntryIdEncoded, TimelineErrorBodyEncoded, TimelineMessageBodyEncoded, TimelinePageEncoded, TimelineRecipientRefEncoded, TimelineRedactionBodyEncoded, TimelineStatusEncoded, TimelineStatusBodyEncoded, TimelineTodoItemEncoded, TimelineToolCallBodyEncoded, TimelineToolResultBodyEncoded, TimelineTruncationBodyEncoded, TimelineUsageBodyEncoded, TimelineViewEncoded, TimelineViewAskEncoded, TimelineViewAssistantErrorEncoded, TimelineViewBashEncoded, TimelineViewCompactionEncoded, TimelineViewCredentialPinEncoded, TimelineViewEditEncoded, TimelineViewEvalEncoded, TimelineViewGenericEncoded, TimelineViewHubEncoded, TimelineViewIrcEncoded, TimelineViewJobEncoded, TimelineViewModelChangeEncoded, TimelineViewOmpEventEncoded, TimelineViewReadEncoded, TimelineViewResetBoundaryEncoded, TimelineViewSearchEncoded, TimelineViewSessionExitEncoded, TimelineViewSkillEncoded, TimelineViewTaskEncoded, TimelineViewThinkingLevelEncoded, TimelineViewTitleEncoded, TimelineViewTodoEncoded, TimelineViewToolStartEncoded, TimelineViewWriteEncoded, TimestampEncoded, UnknownResourceEncoded, UsageEncoded, UsageAccountEncoded, UsageContextEncoded, UsageCreditsEncoded, UsageLimitEncoded, UsagePeriodEncoded, UsagePricingEncoded, UsagePricingRatesEncoded, UsageRowEncoded, UsageSummaryEncoded, ValueEncoded, VisualizationEncoded, VisualizationEdgeEncoded, VisualizationGroupEncoded, VisualizationNodeEncoded, WorkEncoded, WorkLabelEncoded, WorkProgramEncoded, WorkStateEncoded } from "./ReadSchema.generated.ts" export * from "./ReadSchema.generated.ts" export const ActionCommon = /*#__PURE__*/ (() => Schema.Struct({ diff --git a/crates/fractal-api-core/src/api.rs b/crates/fractal-api-core/src/api.rs index 6c08391af1d..439a96a43a1 100644 --- a/crates/fractal-api-core/src/api.rs +++ b/crates/fractal-api-core/src/api.rs @@ -2942,11 +2942,13 @@ fn client_work_values( .into_iter() .flat_map(|summaries| summaries.iter().copied()), ); + let spec = work + .agentless + .then(|| step_specs.get(&work.run).and_then(|steps| steps.get(&work.step))) + .flatten(); let gate_kind = if work.agentless { - let spec = step_specs - .get(&work.run) - .and_then(|steps| steps.get(&work.step)); match spec { + Some(spec) if spec.program.is_some() => Some("program"), Some(spec) if spec.after_run.is_some() => Some("run"), Some(spec) if spec.gates.is_empty() => Some("watch"), Some(spec) @@ -2990,6 +2992,21 @@ fn client_work_values( } else { None }; + // What a program step runs now: its innermost open key, from the newest entries. + let program = spec + .and_then(|spec| spec.program.as_ref()) + .map(|program| -> anyhow::Result { + let journal = store.program_journal(&work.subject, work.attempt, crate::program_journal::SUMMARY_ENTRIES)?; + let open = journal.innermost(); + Ok(json!({ + "open_key": open.map(|entry| &entry.key), + "open_since_unix_ms": open.and_then(|entry| entry.started_at_unix_ms), + "child": open.and_then(|entry| entry.child.as_ref()), + "last_entry_at_unix_ms": journal.last_entry_at_unix_ms, + "stalled_after_ms": crate::program_journal::stalled_after_ms(program), + })) + }) + .transpose()?; Ok(json!({ "id": work.subject, "kind": "work", @@ -3007,6 +3024,7 @@ fn client_work_values( "state": state, "agentless": work.agentless, "gate_kind": gate_kind, + "program": program, "attempt": work.attempt, "readiness_epoch": work.readiness_epoch, "claimant": work.claimant, diff --git a/crates/fractal-api-core/src/api/waiting_on.rs b/crates/fractal-api-core/src/api/waiting_on.rs index 0997bcbccbc..edc8a00db9e 100644 --- a/crates/fractal-api-core/src/api/waiting_on.rs +++ b/crates/fractal-api-core/src/api/waiting_on.rs @@ -132,6 +132,13 @@ fn classify(row: &WorkRow, context: &Context<'_>) -> StepWait { wait.detail = ask .and_then(|ask| ask.view.title.clone().or_else(|| ask.view.goals.first().cloned())) .or_else(|| view.blocked_reason.clone()); + // A program asks under one of its keys: name it. + if let Some(key) = row.value["program"]["open_key"].as_str() { + wait.detail = Some(match wait.detail.take() { + Some(detail) => format!("{key}: {detail}"), + None => key.to_owned(), + }); + } return wait; } if view.status == "blocked" { @@ -145,6 +152,7 @@ fn classify(row: &WorkRow, context: &Context<'_>) -> StepWait { if view.agentless { wait.who = Some("st".into()); match row.value["gate_kind"].as_str() { + Some("program") => classify_program(&mut wait, view, &row.value["program"], context.now), Some("watch") => wait.kind = WaitKind::Watching, Some("run") => { wait.kind = WaitKind::Hold; @@ -240,6 +248,34 @@ fn classify(row: &WorkRow, context: &Context<'_>) -> StepWait { wait } +/// A program step moves while its journal does: past its `stalled-after` without a new entry it +/// has made no progress. The detail names the innermost open key and what it waits on. +fn classify_program(wait: &mut StepWait, view: &StepRunView, program: &Value, now: u128) { + let at = |name: &str| program[name].as_u64().map(u128::from); + let since = [at("last_entry_at_unix_ms"), view.execution_started_at_unix_ms] + .into_iter() + .flatten() + .max() + .unwrap_or(view.updated_at_unix_ms); + let stalled_after = program["stalled_after_ms"] + .as_u64() + .unwrap_or(crate::program_journal::DEFAULT_STALLED_AFTER_MS); + wait.since_unix_ms = Some(since); + wait.kind = if now.saturating_sub(since) > u128::from(stalled_after) { + WaitKind::NoProgress + } else { + WaitKind::Program + }; + wait.detail = Some(match (program["open_key"].as_str(), program["child"].as_str()) { + (Some(key), Some(child)) => format!("{key} waits on {child}"), + (Some(key), None) => match at("open_since_unix_ms") { + Some(started) => format!("{key} running for {}", describe_age(now.saturating_sub(started))), + None => format!("{key} running"), + }, + (None, _) => "between steps".to_owned(), + }); +} + /// Most blocking first: the causes in their order, a step that asked before the ask it made, /// then whatever has waited longest. fn sort_waits(waits: &mut [StepWait]) { @@ -288,6 +324,7 @@ pub fn describe(wait: &StepWait, now: u128) -> String { _ => format!("{}: held{detail}", wait.path), }, WaitKind::Working => format!("{}: {who} working, last progress {age} ago{detail}", wait.path), + WaitKind::Program => format!("{}: program, last journal entry {age} ago{detail}", wait.path), WaitKind::Verifying => format!("{}: verifying{detail}", wait.path), WaitKind::Ready => format!("{}: ready for {age}", wait.path), WaitKind::Queued => format!("{}: queued for {who}{detail}", wait.path), diff --git a/crates/fractal-api-core/src/api/waiting_on_tests.rs b/crates/fractal-api-core/src/api/waiting_on_tests.rs index 021af844a93..57c228393cf 100644 --- a/crates/fractal-api-core/src/api/waiting_on_tests.rs +++ b/crates/fractal-api-core/src/api/waiting_on_tests.rs @@ -253,6 +253,93 @@ fn a_stopped_seat_is_the_cause_of_the_step_it_would_take() { assert_eq!(blocked.groups.iter().map(|group| group.cause).collect::>(), [WaitKind::SeatDown]); } +#[test] +fn a_program_step_names_its_innermost_open_key_and_stalls_past_its_stalled_after() { + use sha2::Digest as _; + let t0 = start_time(); + set_thread_clock(Some(t0)); + let _clock = Clock; + let root = tempfile::tempdir().unwrap(); + let store = Store::open(&root.path().join("graph.db"), "cedar").unwrap(); + let bundle = b"export default {}"; + let hash = hex::encode(sha2::Sha256::digest(bundle)); + store.put_blob(bundle).unwrap(); + publish( + &store, + &format!( + r#"version 2 +mission "garden/program" state="ready" {{ + goal "Grow the garden by program." + step "land" {{ program "/bundles/land.mjs" sha256="{hash}" stalled-after="10m" }} +}}"# + ), + "publish-garden-program", + ); + let run = start(&store, "garden/program", "program-1"); + let land = run.steps.iter().find(|step| step.step == "land").unwrap().subject.clone(); + store.set_step_state(&land, "working", None).unwrap(); + let entry = |key: &str, status: &str, child: Option<&str>| { + let mut fields = BTreeMap::from([ + ("key".to_owned(), json!(key)), + ("status".to_owned(), json!(status)), + ("attempt".to_owned(), json!(1)), + ("program".to_owned(), json!(hash)), + ]); + if let Some(child) = child { + fields.insert("child".to_owned(), json!(child)); + } + store + .append_claim(&crate::model::ClaimInput { + subject: land.clone(), + kind: "program.step".into(), + actor: None, + fields, + evidence: Vec::new(), + expected_subject: None, + idempotency_key: None, + }) + .unwrap(); + }; + // The store stamps entries with its own clock, a moment after `t0`: read half a minute past + // the whole minute. + entry("render", "started", None); + entry("render", "completed", None); + entry("review", "started", Some("mission-run/garden/child")); + + // The journal moved a minute ago: the program runs, and its innermost key waits on a child. + let waiting = waits(&store, &run.subject, t0 + MINUTE + MINUTE / 2, QUIET); + let (work, _) = published(&store); + assert_eq!(work.rows.rows[&land].value["gate_kind"], "program"); + let landing = open(&waiting, "land"); + assert_eq!(landing.kind, WaitKind::Program, "a program step is never `watching`: {waiting:#?}"); + assert_eq!(landing.detail.as_deref(), Some("review waits on mission-run/garden/child")); + assert_eq!(waiting.verdict, RunVerdict::Moving); + assert_eq!( + waiting.reason, + "land: program, last journal entry 1m ago: review waits on mission-run/garden/child" + ); + + // A running effect is named with how long it has run. + entry("review", "completed", Some("mission-run/garden/child")); + entry("merge", "started", None); + let waiting = waits(&store, &run.subject, t0 + 5 * MINUTE + MINUTE / 2, QUIET); + assert_eq!(open(&waiting, "land").detail.as_deref(), Some("merge running for 5m")); + + // Past its stalled-after with no new entry, well inside the quiet limit: no progress. + let waiting = waits(&store, &run.subject, t0 + 14 * MINUTE + MINUTE / 2, QUIET); + assert_eq!((waiting.verdict, waiting.cause), (RunVerdict::Stalled, Some(WaitKind::NoProgress))); + assert!(waiting.reason.starts_with("land: st holds it, no progress for 14m: merge running for 14m"), "{}", waiting.reason); + + // `missions show` projects the same journal under the step. + let view = store.mission_run(&run.id).unwrap().unwrap(); + let program = &view.programs[0]; + assert_eq!((program.path.as_str(), program.stalled_after_ms, program.fault.as_ref()), ("land", 600_000, None)); + let keys = program.journal.entries.iter().map(|entry| (entry.key.as_str(), entry.status.as_str())).collect::>(); + assert_eq!(keys, [("render", "completed"), ("review", "completed"), ("merge", "started")]); + assert_eq!(program.journal.entries[1].child.as_deref(), Some("mission-run/garden/child")); + assert_eq!(program.journal.open, ["merge"]); +} + #[tokio::test] async fn the_reads_serve_only_published_lists() { let root = tempfile::tempdir().unwrap(); diff --git a/crates/fractal-client-codegen/templates/Models.swift.in b/crates/fractal-client-codegen/templates/Models.swift.in index f67d5ffb1f3..1fd1bfe401b 100644 --- a/crates/fractal-client-codegen/templates/Models.swift.in +++ b/crates/fractal-client-codegen/templates/Models.swift.in @@ -615,7 +615,9 @@ public struct AgentPresentation: Codable, Sendable { public let state, group: St public struct AgentPresentationWork: Codable, Sendable { public let step, mission, path, label: String; public let title: String? } public struct AgentPresentationNow: Codable, Sendable { public let step, mission, path, label, since: String; public let title, goal, holderDownSince: String?; enum CodingKeys: String, CodingKey { case step, mission, path, label, since, title, goal, holderDownSince = "holder_down_since" } } public struct WorkLabel: Codable, Sendable { public let id, missionID, missionRunID, path, state, since: String; public let title, goal, holderDownSince: String?; enum CodingKeys: String, CodingKey { case id, missionID = "mission_id", missionRunID = "mission_run_id", path, title, goal, state, since, holderDownSince = "holder_down_since" } } -public struct WorkResource: Codable, Sendable { public let missionID: String?; public let id, kind, revision, updatedAt, missionRunID, generationID, definitionID, path, state: String; public let gateKind, title, assignedTo, lastProgress, progressAt: String?; public let attempt: UInt; public let readinessEpoch: UInt64; public let claimant, claimIncarnation, submittedIncarnation, blockedReason: String?; public let agentless: Bool?; public let claimExpiresAtUnixMS, executionStartedAtUnixMS, nudgedAtUnixMS, executionElapsedMS, executionWaitedMS, timeoutMS: UInt64?; public let blockers, goals, constraints: [String]; public let personAnswers: [PersonAnswerRecord]?; public let usage: UsageSummary?; public let operational: Operational?; enum CodingKeys: String, CodingKey { case id, kind, revision, updatedAt = "updated_at", missionID = "mission_id", missionRunID = "mission_run_id", generationID = "generation_id", definitionID = "definition_id", path, state, gateKind = "gate_kind", title, assignedTo = "assigned_to", lastProgress = "last_progress", progressAt = "progress_at", attempt, readinessEpoch = "readiness_epoch", claimant, claimIncarnation = "claim_incarnation", submittedIncarnation = "submitted_incarnation", agentless, claimExpiresAtUnixMS = "claim_expires_at_unix_ms", executionStartedAtUnixMS = "execution_started_at_unix_ms", nudgedAtUnixMS = "nudged_at_unix_ms", executionElapsedMS = "execution_elapsed_ms", executionWaitedMS = "execution_waited_ms", timeoutMS = "timeout_ms", blockedReason = "blocked_reason", blockers, goals, constraints, personAnswers = "person_answers", usage, operational } } +public struct WorkResource: Codable, Sendable { public let missionID: String?; public let id, kind, revision, updatedAt, missionRunID, generationID, definitionID, path, state: String; public let gateKind, title, assignedTo, lastProgress, progressAt: String?; public let program: WorkProgram?; public let attempt: UInt; public let readinessEpoch: UInt64; public let claimant, claimIncarnation, submittedIncarnation, blockedReason: String?; public let agentless: Bool?; public let claimExpiresAtUnixMS, executionStartedAtUnixMS, nudgedAtUnixMS, executionElapsedMS, executionWaitedMS, timeoutMS: UInt64?; public let blockers, goals, constraints: [String]; public let personAnswers: [PersonAnswerRecord]?; public let usage: UsageSummary?; public let operational: Operational?; enum CodingKeys: String, CodingKey { case id, kind, revision, updatedAt = "updated_at", missionID = "mission_id", missionRunID = "mission_run_id", generationID = "generation_id", definitionID = "definition_id", path, state, gateKind = "gate_kind", program, title, assignedTo = "assigned_to", lastProgress = "last_progress", progressAt = "progress_at", attempt, readinessEpoch = "readiness_epoch", claimant, claimIncarnation = "claim_incarnation", submittedIncarnation = "submitted_incarnation", agentless, claimExpiresAtUnixMS = "claim_expires_at_unix_ms", executionStartedAtUnixMS = "execution_started_at_unix_ms", nudgedAtUnixMS = "nudged_at_unix_ms", executionElapsedMS = "execution_elapsed_ms", executionWaitedMS = "execution_waited_ms", timeoutMS = "timeout_ms", blockedReason = "blocked_reason", blockers, goals, constraints, personAnswers = "person_answers", usage, operational } } +/// A program step's innermost open journal key and journal timing. +public struct WorkProgram: Codable, Sendable { public let openKey, child: String?; public let openSinceUnixMS, lastEntryAtUnixMS: UInt64?; public let stalledAfterMS: UInt64; enum CodingKeys: String, CodingKey { case openKey = "open_key", child, openSinceUnixMS = "open_since_unix_ms", lastEntryAtUnixMS = "last_entry_at_unix_ms", stalledAfterMS = "stalled_after_ms" } } public struct AgentRelationship: Codable, Sendable { public let agentID: String; public let reason: String?; enum CodingKeys: String, CodingKey { case agentID = "agent_id", reason } } public enum ResourceReferenceKind: String, Codable, Sendable { case uri, subject } public enum ResourceReferenceResolution: String, Codable, Sendable { case resolved, unresolved; case notApplicable = "not-applicable" } diff --git a/crates/fractal-client-codegen/templates/generated.rs.in b/crates/fractal-client-codegen/templates/generated.rs.in index a2a87f21f7d..34f6879cc03 100644 --- a/crates/fractal-client-codegen/templates/generated.rs.in +++ b/crates/fractal-client-codegen/templates/generated.rs.in @@ -1828,6 +1828,9 @@ pub struct Work { pub progress_at: Option, pub state: String, pub gate_kind: Option, + /// A program step's innermost open journal key and journal timing. + #[serde(default)] + pub program: Option, pub attempt: u64, pub readiness_epoch: u64, pub claimant: Option, @@ -1863,6 +1866,19 @@ pub struct Work { #[serde(default)] pub usage: Option, } +/// A program step's innermost open journal key and journal timing. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] +pub struct WorkProgram { + #[serde(default)] + pub open_key: Option, + #[serde(default)] + pub open_since_unix_ms: Option, + #[serde(default)] + pub child: Option, + #[serde(default)] + pub last_entry_at_unix_ms: Option, + pub stalled_after_ms: u64, +} #[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] pub struct AgentRelationship { pub agent_id: String, diff --git a/crates/fractal-client/src/generated.rs b/crates/fractal-client/src/generated.rs index c26061f4283..912535e5d51 100644 --- a/crates/fractal-client/src/generated.rs +++ b/crates/fractal-client/src/generated.rs @@ -1829,6 +1829,9 @@ pub struct Work { pub progress_at: Option, pub state: String, pub gate_kind: Option, + /// A program step's innermost open journal key and journal timing. + #[serde(default)] + pub program: Option, pub attempt: u64, pub readiness_epoch: u64, pub claimant: Option, @@ -1864,6 +1867,19 @@ pub struct Work { #[serde(default)] pub usage: Option, } +/// A program step's innermost open journal key and journal timing. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] +pub struct WorkProgram { + #[serde(default)] + pub open_key: Option, + #[serde(default)] + pub open_since_unix_ms: Option, + #[serde(default)] + pub child: Option, + #[serde(default)] + pub last_entry_at_unix_ms: Option, + pub stalled_after_ms: u64, +} #[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] pub struct AgentRelationship { pub agent_id: String, diff --git a/crates/fractal-daemon/src/lib.rs b/crates/fractal-daemon/src/lib.rs index 6119b3c72b8..b53c0458211 100644 --- a/crates/fractal-daemon/src/lib.rs +++ b/crates/fractal-daemon/src/lib.rs @@ -135,6 +135,7 @@ mod telemetry_tests; /// Attaches a local terminal to another fleet host's PTY session over Fabric, without st daemons. pub mod terminal_fabric; pub use fractal_kernel::terminal_binding; +pub use fractal_kernel::program_journal; pub use fractal_kernel::waiting_on; /// Runs each agent as its own Unix user, in its own namespaces and root-owned cgroup. pub use fractal_store::tile_registry; diff --git a/crates/fractal-kernel/src/lib.rs b/crates/fractal-kernel/src/lib.rs index eaf1517928a..113c21b53d9 100644 --- a/crates/fractal-kernel/src/lib.rs +++ b/crates/fractal-kernel/src/lib.rs @@ -28,6 +28,8 @@ pub mod memory; pub mod read_cache; pub mod model; pub mod pricing; +/// What a program step's journal shows: each inner step's latest status, timing and failure. +pub mod program_journal; pub mod projection; pub mod rules; pub mod seat_queue; diff --git a/crates/fractal-kernel/src/model.rs b/crates/fractal-kernel/src/model.rs index 334845ea1cc..57fce511ca1 100644 --- a/crates/fractal-kernel/src/model.rs +++ b/crates/fractal-kernel/src/model.rs @@ -3030,6 +3030,9 @@ pub struct MissionRunView { pub steps: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub loops: Vec, + /// Each program step's current journal, computed for presentation. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub programs: Vec, /// Unresolved exit-code field gates on terminal execs, computed for mission details. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub stuck_gates: Vec, diff --git a/crates/fractal-kernel/src/program_journal.rs b/crates/fractal-kernel/src/program_journal.rs new file mode 100644 index 00000000000..79c7db420d1 --- /dev/null +++ b/crates/fractal-kernel/src/program_journal.rs @@ -0,0 +1,294 @@ +//! What a program step's journal shows (docs/st3/program-engine.md §8): each inner step's latest +//! status, timing, failure and linked child, folded from the step's newest `program.step` claims. + +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; + +/// The claim kind of a program journal entry, on the program's `step-run` subject. +pub const JOURNAL_KIND: &str = "program.step"; +/// The reserved journal key the daemon writes on `done` with the program's encoded success value. +/// It is the program's result, not an inner step: replay and views skip it. +pub const RESULT_KEY: &str = "$result"; +/// The fault on a program step whose journal shows no new entry within its `stalled-after`. +pub const STALLED_FAULT_SCOPE: &str = "program/stalled"; +/// The fault on a program step whose host cannot run it, or whose worker stopped answering. +pub const HOST_FAULT_SCOPE: &str = "program/host"; +/// How long a program step may go without a journal entry when it declares no `stalled-after`: +/// the mission-level default. +pub const DEFAULT_STALLED_AFTER_MS: u64 = 30 * 60 * 1000; +/// How many of a step's newest journal entries a view folds. A long loop's older keys are +/// counted out as `truncated`, never replayed for presentation. +pub const VIEW_ENTRIES: usize = 200; +/// How many of a step's newest journal entries the work list and the stall check read to name +/// the innermost open key. +pub const SUMMARY_ENTRIES: usize = 64; +/// How much of an entry's typed error a view keeps. +const ERROR_CHARS: usize = 300; + +/// How long `program` may go without a new journal entry before it counts as stalled. +pub fn stalled_after_ms(program: &crate::model::ProgramSpec) -> u64 { + program.stalled_after_ms.unwrap_or(DEFAULT_STALLED_AFTER_MS) +} + +/// One inner step: the latest entry of its key in the current attempt. +#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)] +pub struct ProgramEntry { + pub key: String, + /// `started`, `completed` or `failed`. + pub status: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub started_at_unix_ms: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ended_at_unix_ms: Option, + /// The effect's duration, as the host measured it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ms: Option, + /// The typed failure, with its `_tag` and fields. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, + /// The child `mission-run/…` or ask subject the key waits on. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub child: Option, +} + +/// One attempt's journal, as far as a view reads it. +#[derive(Clone, Debug, Default, Deserialize, PartialEq, Eq, Serialize)] +pub struct ProgramJournal { + /// Each key's latest entry, in the order the keys first appear. + #[serde(default)] + pub entries: Vec, + /// The keys whose latest entry is `started`, most recently started first: the first is the + /// innermost open key. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub open: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_entry_at_unix_ms: Option, + /// Older entries exist beyond the ones read. + #[serde(default)] + pub truncated: bool, +} + +impl ProgramJournal { + /// Fold one attempt's entries, newest first: each claim's `fields` and acceptance time. + pub fn fold<'a>(newest_first: impl IntoIterator, truncated: bool) -> Self { + let mut journal = Self { truncated, ..Self::default() }; + // Newest first, so a key's first entry seen is its latest, and the last one seen orders it. + let mut index = HashMap::<&str, usize>::new(); + let mut keys = Vec::<(usize, ProgramEntry)>::new(); + for (seen, (fields, at)) in newest_first.into_iter().enumerate() { + journal.last_entry_at_unix_ms.get_or_insert(at); + let Some(key) = fields.get("key").and_then(Value::as_str).filter(|key| *key != RESULT_KEY) else { continue }; + let status = fields.get("status").and_then(Value::as_str).unwrap_or_default(); + let slot = *index.entry(key).or_insert_with(|| { + keys.push(( + seen, + ProgramEntry { + key: key.to_owned(), + status: status.to_owned(), + started_at_unix_ms: None, + ended_at_unix_ms: (status != "started").then_some(at), + ms: fields.get("ms").and_then(Value::as_u64), + error: fields.get("error").filter(|error| !error.is_null()).map(error_text), + child: None, + }, + )); + keys.len() - 1 + }); + let (oldest, entry) = &mut keys[slot]; + *oldest = seen; + // A resumed effect starts again: its latest start is the one that runs. + if status == "started" && entry.started_at_unix_ms.is_none() { + entry.started_at_unix_ms = Some(at); + } + if entry.child.is_none() { + entry.child = fields.get("child").and_then(Value::as_str).map(str::to_owned); + } + } + keys.sort_by_key(|(oldest, _)| std::cmp::Reverse(*oldest)); + let mut open = keys + .iter() + .filter(|(_, entry)| entry.status == "started") + .map(|(_, entry)| (entry.started_at_unix_ms, entry.key.clone())) + .collect::>(); + open.sort_by(|left, right| right.cmp(left)); + journal.open = open.into_iter().map(|(_, key)| key).collect(); + journal.entries = keys.into_iter().map(|(_, entry)| entry).collect(); + journal + } + + /// The innermost open key's entry. + pub fn innermost(&self) -> Option<&ProgramEntry> { + let key = self.open.first()?; + self.entries.iter().find(|entry| &entry.key == key) + } +} + +/// A program step of a run, with its current attempt's journal, as `missions show` and the run +/// view present it. The journal's fields sit at the top level. +/// +/// Deserializing goes through `ProgramRunViewFields`: serde's `flatten` buffers the journal +/// through a value type that cannot hold its `u128` timestamps. +#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)] +#[serde(from = "ProgramRunViewFields")] +pub struct ProgramRunView { + pub step_run: String, + pub path: String, + pub attempt: u32, + pub bundle_hash: String, + /// The declared `stalled-after`, else [`DEFAULT_STALLED_AFTER_MS`]. + pub stalled_after_ms: u64, + #[serde(flatten)] + pub journal: ProgramJournal, + /// The open `program/stalled` or `program/host` fault on the step. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub fault: Option, +} + +/// [`ProgramRunView`]'s wire fields, with the journal's spelled out. +#[derive(Deserialize)] +struct ProgramRunViewFields { + step_run: String, + path: String, + attempt: u32, + bundle_hash: String, + stalled_after_ms: u64, + #[serde(default)] + entries: Vec, + #[serde(default)] + open: Vec, + #[serde(default)] + last_entry_at_unix_ms: Option, + #[serde(default)] + truncated: bool, + #[serde(default)] + fault: Option, +} + +impl From for ProgramRunView { + fn from(fields: ProgramRunViewFields) -> Self { + let ProgramRunViewFields { step_run, path, attempt, bundle_hash, stalled_after_ms, entries, open, last_entry_at_unix_ms, truncated, fault } = fields; + Self { + step_run, + path, + attempt, + bundle_hash, + stalled_after_ms, + journal: ProgramJournal { entries, open, last_entry_at_unix_ms, truncated }, + fault, + } + } +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)] +pub struct ProgramFault { + pub scope: String, + pub reason: String, +} + +/// A typed failure keeps its tag and fields; a long one is cut. +fn error_text(error: &Value) -> String { + let text = match error { + Value::String(text) => text.clone(), + other => other.to_string(), + }; + if text.chars().count() <= ERROR_CHARS { + return text; + } + let mut cut = text.chars().take(ERROR_CHARS).collect::(); + cut.push('…'); + cut +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn a_journal_shows_each_keys_latest_entry_and_its_innermost_open_key() { + // Newest first, as the store reads them. + let rows = [ + (json!({"key": "review", "status": "started", "child": "mission-run/child"}), 60), + (json!({"key": "check", "status": "started"}), 50), + (json!({"key": "merge", "status": "failed", "ms": 900, "error": {"_tag": "Conflict", "path": "a.rs"}}), 40), + (json!({"key": "merge", "status": "started"}), 30), + (json!({"key": "render", "status": "completed", "ms": 120, "value": "ok"}), 20), + (json!({"key": "render", "status": "started"}), 10), + ]; + let journal = ProgramJournal::fold(rows.iter().map(|(fields, at)| (fields, *at)), false); + let keys = journal.entries.iter().map(|entry| (entry.key.as_str(), entry.status.as_str())).collect::>(); + assert_eq!(keys, [("render", "completed"), ("merge", "failed"), ("check", "started"), ("review", "started")]); + assert_eq!(journal.open, ["review", "check"]); + assert_eq!(journal.last_entry_at_unix_ms, Some(60)); + let render = &journal.entries[0]; + assert_eq!((render.started_at_unix_ms, render.ended_at_unix_ms, render.ms), (Some(10), Some(20), Some(120))); + assert_eq!(journal.entries[1].error.as_deref(), Some(r#"{"_tag":"Conflict","path":"a.rs"}"#)); + let innermost = journal.innermost().unwrap(); + assert_eq!((innermost.key.as_str(), innermost.child.as_deref()), ("review", Some("mission-run/child"))); + assert_eq!(innermost.ended_at_unix_ms, None); + } + + #[test] + fn the_program_result_is_not_an_inner_step() { + let rows = [ + (json!({"key": RESULT_KEY, "status": "completed", "value": {"merged": true}}), 30), + (json!({"key": "merge", "status": "completed", "ms": 40}), 20), + (json!({"key": "merge", "status": "started"}), 10), + ]; + let journal = ProgramJournal::fold(rows.iter().map(|(fields, at)| (fields, *at)), false); + let keys = journal.entries.iter().map(|entry| entry.key.as_str()).collect::>(); + assert_eq!(keys, ["merge"]); + assert_eq!(journal.last_entry_at_unix_ms, Some(30)); + } + + #[test] + fn a_long_error_is_cut() { + let rows = [(json!({"key": "k", "status": "failed", "error": "x".repeat(400)}), 1)]; + let journal = ProgramJournal::fold(rows.iter().map(|(fields, at)| (fields, *at)), true); + let error = journal.entries[0].error.as_deref().unwrap(); + assert_eq!(error.chars().count(), ERROR_CHARS + 1); + assert!(journal.truncated && journal.open.is_empty()); + } + + fn timed_view() -> ProgramRunView { + let rows = [ + (json!({"key": "merge", "status": "started", "child": "mission-run/child"}), 30), + (json!({"key": "render", "status": "completed", "ms": 120}), 20), + (json!({"key": "render", "status": "started"}), 10), + ]; + ProgramRunView { + step_run: "step-run/g/run".into(), + path: "run".into(), + attempt: 1, + bundle_hash: "0".repeat(64), + stalled_after_ms: DEFAULT_STALLED_AFTER_MS, + journal: ProgramJournal::fold(rows.iter().map(|(fields, at)| (fields, *at)), false), + fault: Some(ProgramFault { scope: STALLED_FAULT_SCOPE.into(), reason: "no entry".into() }), + } + } + + #[test] + fn a_program_run_view_keeps_its_journal_fields_at_the_top_level_and_round_trips() { + let view = timed_view(); + let value = serde_json::to_value(&view).unwrap(); + assert_eq!(value["last_entry_at_unix_ms"], 30); + assert_eq!(value["open"], json!(["merge"])); + assert_eq!(value["entries"][0]["started_at_unix_ms"], 10); + assert!(value.get("journal").is_none()); + assert_eq!(serde_json::from_value::(value).unwrap(), view); + } + + #[test] + fn a_mission_run_page_with_a_timed_program_step_round_trips() { + let run = json!({ + "subject": "mission-run/g", "id": "g", "mission": "m", "generation": "g", + "initial_revision": "1", "revision": "2", "root_revision": "1", "root_mission_run": "mission-run/g", + "workspace": "w", "requester": "person/owner", "mode": "run", "status": "running", "phase": "work", + "created_at_unix_ms": 1, "updated_at_unix_ms": 2, "programs": [timed_view()], + }); + let page = json!({"runs": [run], "has_more": false, "frontier": 1}); + let decoded = serde_json::from_value::(page).unwrap(); + assert_eq!(decoded.runs[0].programs, [timed_view()]); + } +} diff --git a/crates/fractal-kernel/src/waiting_on.rs b/crates/fractal-kernel/src/waiting_on.rs index 9c68dc1dfdf..134a085983b 100644 --- a/crates/fractal-kernel/src/waiting_on.rs +++ b/crates/fractal-kernel/src/waiting_on.rs @@ -13,7 +13,8 @@ pub enum WaitKind { Ask, /// The seat that holds or would take the step is stopped, failed, suspended, down or retired. SeatDown, - /// A seat holds the step and has recorded no progress for longer than the quiet limit. + /// A seat holds the step and has recorded no progress for longer than the quiet limit, or a + /// program step's journal has no new entry within its `stalled-after`. NoProgress, /// Ready for longer than the quiet limit, and no seat has claimed it. Unclaimed, @@ -22,6 +23,9 @@ pub enum WaitKind { Hold, /// A seat holds it and has recorded progress within the quiet limit. Working, + /// A program step whose journal moved within its `stalled-after`; the detail names its + /// innermost open key. + Program, /// Submitted, and st is checking its gates. Verifying, /// Ready within the quiet limit; nobody has claimed it yet. @@ -59,6 +63,7 @@ impl WaitKind { WaitKind::Unclaimed => "unclaimed", WaitKind::Hold => "hold", WaitKind::Working => "working", + WaitKind::Program => "program", WaitKind::Verifying => "verifying", WaitKind::Ready => "ready", WaitKind::Queued => "queued", @@ -75,6 +80,7 @@ impl WaitKind { WaitKind::Unclaimed => "ready, nobody claims it", WaitKind::Hold => "held", WaitKind::Working => "working", + WaitKind::Program => "program running", WaitKind::Verifying => "verifying", WaitKind::Ready => "ready", WaitKind::Queued => "queued", diff --git a/crates/fractal-missions/src/store/missions.rs b/crates/fractal-missions/src/store/missions.rs index c5c84bfbdc6..af56b68444b 100644 --- a/crates/fractal-missions/src/store/missions.rs +++ b/crates/fractal-missions/src/store/missions.rs @@ -1086,9 +1086,12 @@ fn mission_root_tree_run_tx(connection: &Connection, id: &str) -> Result rusqlite::Result<()> { +) -> rusqlite::Result> { if view.steps.is_empty() { - return Ok(()); + return Ok(None); } let Some(mission) = pinned_generation_mission_tx(connection, &view.generation)? else { - return Ok(()); + return Ok(None); }; for view in &mut view.steps { if let Some(step) = crate::mission::find_step(&mission, &view.step) { @@ -1331,7 +1335,7 @@ fn enrich_run_step_queues_tx( view.queue_position = step.queue_position; } } - Ok(()) + Ok(Some(mission)) } /// A run's header and each step's effective state, and with `summaries` each step's latest diff --git a/crates/fractal-reconcile/src/reconcile.rs b/crates/fractal-reconcile/src/reconcile.rs index 3f753cdb4b3..7583734c46f 100644 --- a/crates/fractal-reconcile/src/reconcile.rs +++ b/crates/fractal-reconcile/src/reconcile.rs @@ -63,6 +63,7 @@ pub mod placement; pub mod run_report; pub mod start_spacing; pub mod program; +mod program_faults; pub mod start_jobs; pub mod seat_admission; mod seat_admission_queue; @@ -10567,7 +10568,9 @@ impl Reconciler { "mission-run" | FIRST_READINESS_FAULT_SCOPE | run_report::REPORT_FAULT_SCOPE => { !active.contains(subject) } - "step" => !active_steps.contains(subject), + "step" | crate::program_journal::STALLED_FAULT_SCOPE | crate::program_journal::HOST_FAULT_SCOPE => { + !active_steps.contains(subject) + } _ => false, }) .map(|((subject, scope), _)| (subject.clone(), scope.clone())) @@ -11604,6 +11607,12 @@ impl Reconciler { Ok(changed) }); changed |= step_changed.unwrap_or(false); + // After the step's evaluation, so a dispatch whose host could not start is seen now. + if let Some(program) = &step.spec.program + && let Err(error) = self.reconcile_program_faults(view, program, now_ms()) + { + eprintln!("st3: program faults for {}: {error:#}", view.subject); + } // Finally work gets its ordinary evaluation first, including starting its gates // and declarations. Once it only waits on evidence, cancellation must not hold // the mission's active-run slot until the step's execution budget expires. @@ -14412,11 +14421,9 @@ impl Reconciler { }; match self.program_engine().dispatch(dispatch)? { program::Dispatched::Running | program::Dispatched::Started => Ok(false), - // The step keeps its state; `ProgramEngine::host_fault` names the reason. - program::Dispatched::HostDown(reason) => { - eprintln!("fractal: program {}: {reason}", view.subject); - Ok(false) - } + // The step keeps its state; `reconcile_program_faults` raises `program/host` from + // `ProgramEngine::host_fault` and dispatches it again at a deadline. + program::Dispatched::HostDown(_) => Ok(false), program::Dispatched::Refused(reason) => self.store.set_step_state(&view.subject, "failed", Some(&reason)), } } @@ -21246,6 +21253,7 @@ pub mod tests { mod seat_admission_queue; mod terminal_step_guard; mod program_engine; + mod program_faults; #[test] fn native_exec_and_gate_shell_resolve_the_declared_path() { use super::{NativeRuntime, RuntimeControl}; diff --git a/crates/fractal-reconcile/src/reconcile/program.rs b/crates/fractal-reconcile/src/reconcile/program.rs index 815817283fe..ee5a30ff1a4 100644 --- a/crates/fractal-reconcile/src/reconcile/program.rs +++ b/crates/fractal-reconcile/src/reconcile/program.rs @@ -32,10 +32,7 @@ use crate::MissionsStore; use crate::model::{ClaimInput, ClaimRecord, FractalError, MissionRunView, ProgramSpec}; use crate::store::Store; -/// The claim kind of a program journal entry, on the program's `step-run` subject. -pub const JOURNAL_KIND: &str = "program.step"; -/// The journal key of a program's own encoded success value, written when it ends. -pub const RESULT_KEY: &str = "$result"; +pub use crate::program_journal::{JOURNAL_KIND, RESULT_KEY}; /// The step states in which the engine runs a program step's current attempt. A step waiting on /// a person keeps its program loaded (§11). pub const RUNNING_STATES: [&str; 4] = ["ready", "working", "blocked", "waiting-person"]; diff --git a/crates/fractal-reconcile/src/reconcile/program_faults.rs b/crates/fractal-reconcile/src/reconcile/program_faults.rs new file mode 100644 index 00000000000..57ce5a54735 --- /dev/null +++ b/crates/fractal-reconcile/src/reconcile/program_faults.rs @@ -0,0 +1,72 @@ +//! A program step's faults (docs/st3/program-engine.md §8), decided while its run is evaluated. +//! +//! - `program/stalled`: a running program step whose journal shows no new entry within its +//! `stalled-after`. The evaluation arms the deadline (`note_due`) instead of scanning; every +//! journal entry is a claim on the step, which its run's evaluation read, so an entry moves the +//! deadline and closes the fault. +//! - `program/host`: the engine reports that no host could run the step, or that its worker +//! missed the liveness deadline (DQ8). A step whose host cannot start keeps its state and is +//! dispatched again at a deadline; the engine wakes the step when it fails it or cannot resume +//! it after a host exit. +//! +//! Both are faults on the step subject, closed when the cause ends or by the inactive sweep. +use super::*; +use crate::program_journal::{self, HOST_FAULT_SCOPE, STALLED_FAULT_SCOPE}; +use anyhow::anyhow; + +/// When a step whose host could not start is dispatched again. +pub const HOST_RETRY_MS: u128 = 30_000; + +impl Reconciler { + /// Record or close the program faults of one program step at `now`. + pub(crate) fn reconcile_program_faults( + &self, + view: &crate::model::StepRunView, + program: &crate::model::ProgramSpec, + now: u128, + ) -> Result<()> { + let host = self.program_engine.get().and_then(|engine| engine.host_fault(&view.subject, view.attempt)); + let running = view.status == "working"; + if let Some(reason) = host { + self.record_fault(&view.subject, HOST_FAULT_SCOPE, Err(anyhow!(reason)))?; + // A step whose host could not start still waits to run: try it again then. + if running { + smallclaims::touched::note_due(now + HOST_RETRY_MS); + } + return self.close_fault(&view.subject, STALLED_FAULT_SCOPE, "the program host fault names why"); + } + self.close_fault(&view.subject, HOST_FAULT_SCOPE, "the program host runs it")?; + if !running { + return self.close_fault(&view.subject, STALLED_FAULT_SCOPE, "the program step is not running"); + } + let journal = self.store.program_journal(&view.subject, view.attempt, program_journal::SUMMARY_ENTRIES)?; + let since = [journal.last_entry_at_unix_ms, view.execution_started_at_unix_ms] + .into_iter() + .flatten() + .max() + .unwrap_or(view.updated_at_unix_ms); + let limit = program_journal::stalled_after_ms(program); + let stalls_at = since + u128::from(limit); + if now < stalls_at { + smallclaims::touched::note_due(stalls_at); + return self.close_fault(&view.subject, STALLED_FAULT_SCOPE, "the program's journal moved"); + } + // The reason names no age, so a fault that lasts is recorded once. + let open = match journal.innermost() { + Some(entry) => match &entry.child { + Some(child) => format!("open key `{}` waits on {child}", entry.key), + None => format!("open key `{}`", entry.key), + }, + None => "no open key".to_owned(), + }; + self.record_fault( + &view.subject, + STALLED_FAULT_SCOPE, + Err(anyhow!( + "program step `{}` wrote no journal entry within its stalled-after of {}; {open}", + view.step, + crate::waiting_on::describe_age(u128::from(limit)) + )), + ) + } +} diff --git a/crates/fractal-reconcile/src/reconcile/tests/program_faults.rs b/crates/fractal-reconcile/src/reconcile/tests/program_faults.rs new file mode 100644 index 00000000000..943d6180895 --- /dev/null +++ b/crates/fractal-reconcile/src/reconcile/tests/program_faults.rs @@ -0,0 +1,147 @@ +//! A program step's faults (docs/st3/program-engine.md §8): `program/stalled` from a deadline the +//! evaluation arms on the journal's last entry, never a scan, and `program/host` from the engine. +use super::*; +use crate::program_journal::{HOST_FAULT_SCOPE, STALLED_FAULT_SCOPE}; +use crate::reconcile::program::{JOURNAL_KIND, Liveness, ProgramEngine}; + +const MINUTE: u128 = 60_000; + +/// A store with one run of a program mission whose step stalls after ten minutes. +fn program_run(store: &Store) -> (crate::model::MissionRunView, String) { + let hash = store.put_blob(b"export default {}").unwrap(); + apply_source( + store, + &format!( + r#"version 2 +mission "program-faults" state="ready" {{ + goal "Run a program." + concurrent-runs max=4 + step "program" {{ program "/bundles/program.mjs" sha256="{hash}" stalled-after="10m" }} +}}"# + ), + "program-faults-definition", + ); + let run = store + .create_mission_run(&MissionRunRequest { + mission: "program-faults".into(), + revision: None, + workspace: "/tmp".into(), + requester: None, + mode: None, + inputs: BTreeMap::new(), + idempotency_key: "program-faults-run".into(), + }) + .unwrap(); + let step = run.steps.iter().find(|step| step.step == "program").unwrap().subject.clone(); + (run, step) +} + +fn entry(store: &Store, step: &str, key: &str, status: &str) { + store + .append_claim(&crate::model::ClaimInput { + subject: step.to_owned(), + kind: JOURNAL_KIND.into(), + actor: None, + fields: BTreeMap::from([ + ("key".to_owned(), serde_json::Value::from(key)), + ("status".to_owned(), serde_json::Value::from(status)), + ("attempt".to_owned(), serde_json::Value::from(1)), + ("program".to_owned(), serde_json::Value::from("a".repeat(64))), + ]), + evidence: Vec::new(), + expected_subject: None, + idempotency_key: None, + }) + .unwrap(); +} + +#[test] +fn a_quiet_journal_raises_program_stalled_at_its_deadline_and_an_entry_closes_it() { + let store = Arc::new(Store::open_memory("node").unwrap()); + let reconciler = Reconciler::new(store.clone(), Arc::new(FakeRuntime::default()), "node".into(), Arc::new(Notify::new())); + let (_, step) = program_run(&store); + store.set_step_state(&step, "working", None).unwrap(); + entry(&store, &step, "render", "started"); + entry(&store, &step, "render", "completed"); + entry(&store, &step, "merge", "started"); + let program = |at: u128| { + let view = store.step_run(&step).unwrap().unwrap(); + let spec = crate::model::ProgramSpec { + bundle: "/bundles/program.mjs".into(), + bundle_hash: "a".repeat(64), + stalled_after_ms: Some(10 * MINUTE as u64), + }; + let ((), due) = smallclaims::touched::record_due(|| reconciler.reconcile_program_faults(&view, &spec, at).unwrap()); + due + }; + let last = store.newest_claim_for_subject_kind(&step, JOURNAL_KIND).unwrap().unwrap().accepted_at_unix_ms; + + // Inside its stalled-after: no fault, and the evaluation is due again exactly at the deadline. + assert_eq!(program(last + MINUTE), Some(last + 10 * MINUTE)); + assert_eq!(store.reconcile_fault(&step, STALLED_FAULT_SCOPE).unwrap(), None); + + // At the deadline the fault names the open key; a later evaluation records nothing new. + assert_eq!(program(last + 10 * MINUTE), None); + let reason = store.reconcile_fault(&step, STALLED_FAULT_SCOPE).unwrap().expect("program/stalled is raised"); + assert_eq!(reason, "program step `program` wrote no journal entry within its stalled-after of 10m; open key `merge`"); + program(last + 20 * MINUTE); + let faulted = |status: &str| { + store + .claims_for(&step, Some("reconcile.fault")) + .unwrap() + .iter() + .filter(|claim| claim.body["fields"]["scope"] == STALLED_FAULT_SCOPE && claim.body["fields"]["status"] == status) + .count() + }; + assert_eq!(faulted("faulted"), 1, "a lasting stall is recorded once"); + + // Progress moves the deadline and closes the fault. + entry(&store, &step, "merge", "completed"); + let moved = store.newest_claim_for_subject_kind(&step, JOURNAL_KIND).unwrap().unwrap().accepted_at_unix_ms; + assert_eq!(program(moved + MINUTE), Some(moved + 10 * MINUTE)); + assert_eq!(store.reconcile_fault(&step, STALLED_FAULT_SCOPE).unwrap(), None); + + // A step that stops running cannot stall: no deadline, and an open stall closes. + program(moved + 10 * MINUTE); + assert!(store.reconcile_fault(&step, STALLED_FAULT_SCOPE).unwrap().is_some()); + store.set_step_state(&step, "completed", None).unwrap(); + assert_eq!(program(moved + 30 * MINUTE), None); + assert_eq!(store.reconcile_fault(&step, STALLED_FAULT_SCOPE).unwrap(), None); +} + +#[test] +fn a_host_that_cannot_start_raises_program_host_until_the_run_ends() { + let dir = tempfile::tempdir().unwrap(); + let store = Arc::new(Store::open_memory("node").unwrap()); + let reconciler = Reconciler::new(store.clone(), Arc::new(FakeRuntime::default()), "node".into(), Arc::new(Notify::new())); + // The engine records effect process groups in `groups`: a file there keeps any host from + // starting, with or without a JavaScript runtime. + let programs = dir.path().join("programs"); + std::fs::create_dir_all(&programs).unwrap(); + std::fs::write(programs.join("groups"), b"").unwrap(); + let engine = ProgramEngine::new(store.clone(), Arc::new(|_: &str| {}), programs, Liveness::default()); + assert!(reconciler.program_engine.set(engine.clone()).is_ok()); + let (run, step) = program_run(&store); + + let mut passes = 0; + while store.reconcile_fault(&step, HOST_FAULT_SCOPE).unwrap().is_none() { + passes += 1; + assert!(passes <= 5, "program/host was not raised"); + reconciler.reconcile_once().unwrap(); + } + let reason = store.reconcile_fault(&step, HOST_FAULT_SCOPE).unwrap().unwrap(); + assert!(reason.starts_with("the program host cannot start"), "{reason}"); + assert_eq!(engine.host_fault(&step, 1).as_deref(), Some(reason.as_str())); + // The step keeps waiting to run; it neither fails nor counts as stalled. + assert_eq!(store.step_run(&step).unwrap().unwrap().status, "working"); + assert_eq!(store.reconcile_fault(&step, STALLED_FAULT_SCOPE).unwrap(), None); + + // Once its run ends, the step has nothing left to fault: the inactive sweep closes it. + store.set_step_state(&step, "completed", None).unwrap(); + let mut passes = 0; + while store.reconcile_fault(&step, HOST_FAULT_SCOPE).unwrap().is_some() { + passes += 1; + assert!(passes <= 10, "program/host stayed open after run {} ended", run.subject); + reconciler.reconcile_once().unwrap(); + } +} diff --git a/crates/fractal-store/src/store.rs b/crates/fractal-store/src/store.rs index 1116beaa253..4bda6c12b0c 100644 --- a/crates/fractal-store/src/store.rs +++ b/crates/fractal-store/src/store.rs @@ -176,6 +176,7 @@ use sha2::{Digest as _, Sha256}; use uuid::Uuid; use crate::model::{ApplyResponse, AttentionActionView, AttentionClosing, AttentionItemView, AttentionRequest, AttentionRequestView, AttentionResolveRequest, AttentionWithdrawRequest, Capability, ClaimInput, ClaimRecord, ClaimsPage, ClaimsSummary, ContextUsage, DependencySpec, DesiredSubject, EventRecord, FaultView, HumanReviewView, IntentInput, LoopRoundView, LoopRunView, MAX_EVAL_TIMEOUT_MS, SESSION_CLAIMS, MessageView, MissionDefinitionView, MissionInputKind, MissionResponse, MissionRevisionOperation, MissionRunDeclaration, MissionRunInput, MissionRunOutcomeView, MissionRunRequest, MissionRunView, MissionSpec, MissionState, NormalizedIntent, OperationalAnnotation, OperationalRepairItem, OperationalRepairPlan, OperationalRepairResult, PlannedAction, PlanningCandidateView, PlanningPreviewView, PlanningSessionDeclaration, PlanningSessionView, PlanningVariantView, ReplicaBatch, ReplicaRepairDeclaration, ResourceRefreshOperation, RevisionCutover, RevisionProposalView, RuntimeResetOperation, FractalError, StatusResponse, StepRunView, SubjectChange, SubjectStatus, UsageSummary, WorkSelector, WorkWakeView}; +use fractal_kernel::program_journal::{self, ProgramFault, ProgramJournal, ProgramRunView}; #[cfg(any(test, feature = "test-support"))] use crate::model::WorkRequest; #[cfg(test)] @@ -3149,32 +3150,7 @@ impl Store { /// The reason for an open fault on `subject` in `scope`, if the latest record is a fault. pub fn reconcile_fault(&self, subject: &str, scope: &str) -> Result> { - let connection = self.readers.get(); - let body: Option = connection - .query_row( - /* sqlite-large-value: allow-for-now lv-hb42a56e4a272 owner=fractal_store::store TODO=trace-statement-and-value-representation */ &canonical_sql( - "SELECT body FROM claims WHERE subject=?1 AND kind='reconcile.fault' - AND json_extract(body, '$.fields.scope')=?2 - ORDER BY CANONICAL_DESC(claims) LIMIT 1", - ), - params![subject, scope], - |row| row.get(0), - ) - .optional()?; - let Some(body) = body else { - return Ok(None); - }; - let body: Value = serde_json::from_str(&body)?; - let fields = body.get("fields").unwrap_or(&body); - Ok( - (fields.get("status").and_then(Value::as_str) == Some("faulted")).then(|| { - fields - .get("reason") - .and_then(Value::as_str) - .unwrap_or_default() - .to_owned() - }), - ) + Ok(reconcile_fault_tx(&self.readers.get(), subject, scope)?) } /// Latest preparation outcome for one message and the recipient's current incarnation. @@ -6847,29 +6823,12 @@ impl Store { descending: bool, limit: usize, ) -> Result { - let connection = self.readers.get(); - let order = if descending { "DESC" } else { "ASC" }; - let query = format!( - "SELECT id, store_index, batch_id, subject, kind, origin, actor, body, predecessors, accepted_at_unix_ms - FROM claims WHERE subject=?1 AND kind=?2 AND (?3 IS NULL OR store_index>>()?; - let next_cursor = if claims.len() > limit { - claims.pop(); - claims.last().map(|claim| claim.store_index) - } else { - None - }; - Ok(ClaimsPage { - claims, - next_cursor, - }) + Ok(claims_for_subject_kind_at_tx(&self.readers.get(), subject, kind, before_index, descending, limit)?) + } + + /// The current attempt's journal of a program step, from its newest `limit` entries. + pub fn program_journal(&self, step: &str, attempt: u32, limit: usize) -> Result { + Ok(program_journal_tx(&self.readers.get(), step, attempt, limit)?) } /// Bounded timeline history for exactly one runtime incarnation. The @@ -7188,7 +7147,7 @@ impl Store { AND (?3 IS NULL OR store_index) -> rusqlite::Result< updated_at_unix_ms: updated.parse().unwrap_or(0), steps: Vec::new(), loops: Vec::new(), + programs: Vec::new(), stuck_gates: Vec::new(), scheduler_fault: None, }) @@ -22543,6 +22508,126 @@ pub fn loop_run_views_tx( Ok(loops) } +/// Each program step of `run` with its current attempt's journal and open program fault. +pub fn program_run_views_tx( + connection: &Connection, + run: &MissionRunView, + mission: &MissionSpec, +) -> rusqlite::Result> { + let mut programs = Vec::new(); + for step in &run.steps { + let Some(program) = crate::mission::find_step(mission, &step.step).and_then(|spec| spec.program.as_ref()) + else { + continue; + }; + let mut fault = None; + for scope in [program_journal::HOST_FAULT_SCOPE, program_journal::STALLED_FAULT_SCOPE] { + if let Some(reason) = reconcile_fault_tx(connection, &step.subject, scope)? { + fault = Some(ProgramFault { scope: scope.to_owned(), reason }); + break; + } + } + programs.push(ProgramRunView { + step_run: step.subject.clone(), + path: step.step.clone(), + attempt: step.attempt, + bundle_hash: program.bundle_hash.clone(), + stalled_after_ms: program_journal::stalled_after_ms(program), + journal: program_journal_tx(connection, &step.subject, step.attempt, program_journal::VIEW_ENTRIES)?, + fault, + }); + } + Ok(programs) +} + +/// One attempt's journal of a program step from the step's newest `limit` entries. Every entry of +/// an earlier attempt is older than the current attempt's first, so the page holds the current +/// attempt's newest entries. +pub fn program_journal_tx( + connection: &Connection, + step: &str, + attempt: u32, + limit: usize, +) -> rusqlite::Result { + let page = claims_for_subject_kind_at_tx(connection, step, program_journal::JOURNAL_KIND, None, true, limit)?; + let current = |claim: &&ClaimRecord| claim.body.pointer("/fields/attempt").and_then(Value::as_u64) == Some(u64::from(attempt)); + let truncated = page.next_cursor.is_some() && page.claims.last().is_some_and(|claim| current(&claim)); + Ok(ProgramJournal::fold( + page.claims + .iter() + .filter(current) + .map(|claim| (claim.body.get("fields").unwrap_or(&claim.body), claim.accepted_at_unix_ms)), + truncated, + )) +} + +/// Bounded claim history for one subject/kind projection, backed by the composite index. +pub fn claims_for_subject_kind_at_tx( + connection: &Connection, + subject: &str, + kind: &str, + before_index: Option, + descending: bool, + limit: usize, +) -> rusqlite::Result { + let order = if descending { "DESC" } else { "ASC" }; + let query = format!( + "SELECT id, store_index, batch_id, subject, kind, origin, actor, body, predecessors, accepted_at_unix_ms + FROM claims WHERE subject=?1 AND kind=?2 AND (?3 IS NULL OR store_index>>()?; + let next_cursor = if claims.len() > limit { + claims.pop(); + claims.last().map(|claim| claim.store_index) + } else { + None + }; + Ok(ClaimsPage { + claims, + next_cursor, + }) +} + +/// The reason for an open fault on `subject` in `scope`, if the latest record is a fault. +pub fn reconcile_fault_tx( + connection: &Connection, + subject: &str, + scope: &str, +) -> rusqlite::Result> { + let body: Option = connection + .query_row( + /* sqlite-large-value: allow-for-now lv-hb42a56e4a272 owner=fractal_store::store TODO=trace-statement-and-value-representation */ &canonical_sql( + "SELECT body FROM claims WHERE subject=?1 AND kind='reconcile.fault' + AND json_extract(body, '$.fields.scope')=?2 + ORDER BY CANONICAL_DESC(claims) LIMIT 1", + ), + params![subject, scope], + |row| row.get(0), + ) + .optional()?; + let Some(body) = body else { + return Ok(None); + }; + let body: Value = serde_json::from_str(&body) + .map_err(|error| rusqlite::Error::FromSqlConversionFailure(body.len(), rusqlite::types::Type::Text, Box::new(error)))?; + let fields = body.get("fields").unwrap_or(&body); + Ok( + (fields.get("status").and_then(Value::as_str) == Some("faulted")).then(|| { + fields + .get("reason") + .and_then(Value::as_str) + .unwrap_or_default() + .to_owned() + }), + ) +} + pub fn revision_proposal_view_tx( connection: &Connection, proposal_id: &str, diff --git a/crates/fractal/src/presentation.rs b/crates/fractal/src/presentation.rs index 6671bce6a75..8249927e9e3 100644 --- a/crates/fractal/src/presentation.rs +++ b/crates/fractal/src/presentation.rs @@ -11,6 +11,7 @@ use fractal_daemon::model::{ AttentionItemView, MissionInputKind, MissionRunView, RevisionCutover, RevisionProposalView, RunGenerationView, StepRunView, }; +use fractal_daemon::program_journal::ProgramRunView; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) struct OutputStyle { @@ -170,6 +171,11 @@ pub(crate) fn render_mission_run( if let Some(fault) = &run.scheduler_fault { let _ = writeln!(output, "SCHEDULER FAULT {fault}"); } + for program in &run.programs { + if let Some(fault) = &program.fault { + let _ = writeln!(output, "PROGRAM FAULT {} ({}): {}", program.path, fault.scope, fault.reason); + } + } } if !selected.inputs.is_empty() { let _ = writeln!(output); @@ -187,7 +193,7 @@ pub(crate) fn render_mission_run( if selected.steps.is_empty() { let _ = writeln!(output, " {}", style.muted("No steps.")); } else { - render_run_steps(&mut output, selected, &children, " ", style); + render_run_steps(&mut output, selected, &children, " ", style, now_unix_ms); } output } @@ -1272,6 +1278,7 @@ fn render_run_steps( children: &BTreeMap<&str, Vec<&MissionRunView>>, indent: &str, style: OutputStyle, + now_unix_ms: u128, ) { let base_depth = run .steps @@ -1327,6 +1334,9 @@ fn render_run_steps( let _ = writeln!(output, "{indent} {reason}"); } } + if let Some(program) = run.programs.iter().find(|program| program.step_run == step.subject) { + render_program(output, program, indent, style, now_unix_ms); + } let prefix = format!("{}/", step.step); let nested = run .steps @@ -1371,12 +1381,73 @@ fn render_run_steps( child.steps.len() ); if !matches!(child.status.as_str(), "completed" | "cancelled") { - render_run_steps(output, child, children, &format!("{indent} "), style); + render_run_steps(output, child, children, &format!("{indent} "), style, now_unix_ms); } } } } +/// How many of a program's newest keys `missions show` lists; `--json` carries every key read. +const PROGRAM_KEYS_SHOWN: usize = 12; + +/// A program step's journal: the innermost open key, then each key's latest status, newest last. +fn render_program( + output: &mut String, + program: &ProgramRunView, + indent: &str, + style: OutputStyle, + now_unix_ms: u128, +) { + let journal = &program.journal; + let current = match journal.innermost() { + Some(entry) => format!("running {}", entry.key), + None => "idle".to_owned(), + }; + let _ = write!(output, "{indent} ↳ program · {current} · {} keys", journal.entries.len()); + if journal.truncated { + let _ = write!(output, "+"); + } + if let Some(at) = journal.last_entry_at_unix_ms { + let _ = write!(output, " · last entry {}", relative_time(at, now_unix_ms)); + } + let _ = writeln!(output); + let hidden = journal.entries.len().saturating_sub(PROGRAM_KEYS_SHOWN); + if hidden > 0 { + let _ = writeln!(output, "{indent} … {hidden} earlier keys"); + } + for entry in &journal.entries[hidden..] { + let mut line = format!( + "{indent} {} {}", + style.status_value(&entry.status, format!("{:<9}", entry.status)), + entry.key + ); + match (entry.ms, entry.started_at_unix_ms) { + (Some(ms), _) => { + let _ = write!(line, " · {}", program_duration(u128::from(ms))); + } + (None, Some(started)) if entry.status == "started" => { + let _ = write!(line, " · running {}", program_duration(now_unix_ms.saturating_sub(started))); + } + _ => {} + } + if let Some(child) = &entry.child { + let _ = write!(line, " · waits on {child}"); + } + if let Some(error) = &entry.error { + let _ = write!(line, " · {}", glance(error)); + } + let _ = writeln!(output, "{line}"); + } +} + +fn program_duration(ms: u128) -> String { + if ms < 1_000 { + format!("{ms}ms") + } else { + fractal_daemon::waiting_on::describe_age(ms) + } +} + fn render_flat_step_tree( output: &mut String, steps: &[StepRunView], @@ -1697,6 +1768,7 @@ mod tests { updated_at_unix_ms: 2_000, steps, loops: Vec::new(), + programs: Vec::new(), stuck_gates: Vec::new(), scheduler_fault: None, } @@ -2004,6 +2076,40 @@ mod tests { assert!(rendered.contains("best quality=0.75")); } + #[test] + fn mission_view_shows_each_program_key_and_the_program_fault() { + use fractal_daemon::program_journal::{ProgramFault, ProgramJournal}; + let program_step = step("step-run/demo-generation/land", "land", "working"); + let mut root = run("mission-run/demo/run", None, vec![program_step]); + let rows = [ + (serde_json::json!({"key": "review", "status": "started", "child": "mission-run/demo/review"}), 2_000), + (serde_json::json!({"key": "merge", "status": "failed", "ms": 900, "error": {"_tag": "Conflict"}}), 1_500), + (serde_json::json!({"key": "merge", "status": "started"}), 1_400), + (serde_json::json!({"key": "render", "status": "completed", "ms": 2_500}), 1_200), + ]; + root.programs.push(ProgramRunView { + step_run: "step-run/demo-generation/land".into(), + path: "land".into(), + attempt: 1, + bundle_hash: "a".repeat(64), + stalled_after_ms: 600_000, + journal: ProgramJournal::fold(rows.iter().map(|(fields, at)| (fields, *at)), false), + fault: Some(ProgramFault { + scope: "program/stalled".into(), + reason: "no journal entry for 10m; open key `review`".into(), + }), + }); + + let rendered = render_mission_run(&root, &[root.clone()], OutputStyle::plain(), 62_000); + + assert!(rendered.contains("PROGRAM FAULT land (program/stalled): no journal entry for 10m; open key `review`\n"), "{rendered}"); + assert!(rendered.contains(" ↳ program · running review · 3 keys · last entry 1m ago\n"), "{rendered}"); + assert!(rendered.contains(" completed render · 2s\n"), "{rendered}"); + assert!(rendered.contains(" failed merge · 900ms · {\"_tag\":\"Conflict\"}\n"), "{rendered}"); + assert!(rendered.contains(" started review · running 1m · waits on mission-run/demo/review\n"), "{rendered}"); + assert!(!rendered.contains("watching"), "{rendered}"); + } + #[test] fn mission_view_shows_one_summary_line_per_reporting_step() { let mut working = step("step-run/demo-generation/build", "build", "working"); diff --git a/crates/fractal/src/waiting_on_cli.rs b/crates/fractal/src/waiting_on_cli.rs index 04c24852965..bd966bbaa17 100644 --- a/crates/fractal/src/waiting_on_cli.rs +++ b/crates/fractal/src/waiting_on_cli.rs @@ -29,6 +29,9 @@ fn step_line(wait: &StepWait, now: u128) -> String { (WaitKind::NoProgress | WaitKind::Working, Some(since)) => { let _ = write!(line, " · last progress {since} ago"); } + (WaitKind::Program, Some(since)) => { + let _ = write!(line, " · last journal entry {since} ago"); + } (WaitKind::Unclaimed | WaitKind::Ready, Some(since)) => { let _ = write!(line, " · ready {since}"); } diff --git a/docs/st3/client-v0/README.md b/docs/st3/client-v0/README.md index 64465f78006..94b5c615c74 100644 --- a/docs/st3/client-v0/README.md +++ b/docs/st3/client-v0/README.md @@ -1337,7 +1337,10 @@ Observer and subscription lists and details are available at `/v1/client/observe `/v1/client/subscriptions`. Each resource includes its normalized specification, current state, and owning run, generation, and step. Agentless work includes `gate_kind`: `watch` for a standing step with no gate, `predicate`, `command`, `llm`, or `human` for a single gate family, `mixed` -for combined families, and `run` for an `after-run` step. +for combined families, `run` for an `after-run` step, and `program` for a program step. A program +step's work also carries `program`: its innermost open journal key (`open_key`, with +`open_since_unix_ms` and the `child` run or ask that key waits on), `last_entry_at_unix_ms`, and +the `stalled_after_ms` past which `missions waiting-on` reports it as `no-progress`. ## Harness-neutral session timeline diff --git a/docs/st3/client-v0/schemas/client-v0.schema.json b/docs/st3/client-v0/schemas/client-v0.schema.json index 3daf13f9002..30e39499d50 100644 --- a/docs/st3/client-v0/schemas/client-v0.schema.json +++ b/docs/st3/client-v0/schemas/client-v0.schema.json @@ -2337,7 +2337,8 @@ "required": ["kind", "mission_run_id", "generation_id", "definition_id", "path", "state", "attempt", "readiness_epoch", "goals", "constraints"], "properties": { "kind": { "const": "work" }, - "gate_kind": { "enum": ["watch", "predicate", "command", "llm", "human", "mixed", "run", null] }, + "gate_kind": { "enum": ["watch", "predicate", "command", "llm", "human", "mixed", "run", "program", null] }, + "program": { "oneOf": [{ "$ref": "#/$defs/WorkProgram" }, { "type": "null" }], "description": "A program step's innermost open journal key and journal timing; null for other work." }, "mission_id": { "oneOf": [{ "$ref": "#/$defs/Id" }, { "type": "null" }], "description": "The mission this work's run belongs to." }, "mission_run_id": { "$ref": "#/$defs/Id" }, "generation_id": { "$ref": "#/$defs/Id" }, @@ -2370,6 +2371,17 @@ } ] }, + "WorkProgram": { + "type": "object", + "required": ["stalled_after_ms"], + "properties": { + "open_key": { "type": ["string", "null"], "description": "The most recently started journal key without a result: the innermost open key." }, + "open_since_unix_ms": { "type": ["integer", "null"], "minimum": 0, "x-st-codec": "epoch-ms" }, + "child": { "type": ["string", "null"], "description": "The child mission run or ask subject the open key waits on." }, + "last_entry_at_unix_ms": { "type": ["integer", "null"], "minimum": 0, "x-st-codec": "epoch-ms" }, + "stalled_after_ms": { "type": "integer", "minimum": 1, "x-st-codec": "duration-ms", "description": "How long the journal may go without a new entry before the step has made no progress." } + } + }, "AgentWorkspace": { "type": "object", "additionalProperties": false, diff --git a/docs/st3/program-engine.md b/docs/st3/program-engine.md index b4f02ec5219..7614f541272 100644 --- a/docs/st3/program-engine.md +++ b/docs/st3/program-engine.md @@ -350,13 +350,28 @@ Choice: a child run. It needs no standing owner and reuses §6. ### 8. Visibility and stall detection -- `missions show` and the run view project `program.step` claims under the program step: each key - with its latest status, duration and failure, and child runs linked by key. -- `missions waiting-on` names the innermost open key: a running effect, a child run or an ask. -- Stalls: a program step whose journal shows no new entry within its `stalled-after` raises the - run fault with scope `program/stalled`, naming the open key. A dead host that the engine cannot - restart raises `program/host` on every program step it held, and so does a program worker the - host terminated for missing its liveness deadline (DQ8), on that worker's step. +- `missions show` (text and `--json`, field `programs`) projects the current attempt's + `program.step` claims under the program step: each key with its latest status, start and end, + duration (`ms`), typed failure, and the child run or ask it waits on (`child`). The view folds a + step's newest 200 entries and marks older ones `truncated`; the text lists the newest 12 keys. + The innermost open key is the most recently started key without a result. +- `missions waiting-on` gives a program step the kind `program` (never `watching`) and names its + innermost open key: a running effect with how long it runs, or the child run or ask it waits on. + The published work row carries that key (`program`, `gate_kind: program`), read from the step's + newest 64 entries when the row is folded. A program step whose journal has no new entry within + its `stalled-after` (default 30 minutes) is `no-progress`, so its run is stalled. +- Stalls: a running program step whose journal shows no new entry within its `stalled-after` + raises a fault on the step with scope `program/stalled`, naming the open key (and the child it + waits on). The run's evaluation arms the deadline (the last entry, or the start of execution, + plus `stalled-after`) instead of scanning; each journal entry is a claim on the step, so it moves + the deadline and closes the fault, as does the step leaving `working`. +- A host that cannot start or be restarted raises `program/host` on every program step it held, + and so does a program worker the host terminated for missing its liveness deadline (DQ8), on + that worker's step. A step whose host could not start keeps its state and is dispatched again 30 + seconds later; the fault closes when a host runs it. While `program/host` is open the step does + not also raise `program/stalled`. +- Both faults close when their run ends (the inactive-fault sweep), and surface as operator + attention like every reconcile fault that lasts. ### 9. One execution model diff --git a/docs/st3/waiting-on.md b/docs/st3/waiting-on.md index 3946317052e..5b92c7678a2 100644 --- a/docs/st3/waiting-on.md +++ b/docs/st3/waiting-on.md @@ -19,8 +19,10 @@ The HTTP reads are `GET /v1/mission-runs/{run}/waiting-on` and `GET /v1/blocked- Neither read folds claims or replays history. They read state Fractal already keeps current: - The published work list: every open step of a current generation, with its effective state, - holder, lease, last progress, declared wait, wake attempts and blockers. Its refresher refolds - only the steps whose claims changed ([operational queries](operational-queries.md)). + holder, lease, last progress, declared wait, wake attempts and blockers, and for a program step + its innermost open journal key, read from its newest 64 journal entries when the row is folded. + Its refresher refolds only the steps whose claims changed ([operational queries](operational-queries.md)), + and each journal entry is a claim on its step. - The published missions list: the open human gates and the agents roster that mission presentations read, which gives each seat's state. - For one run, one primary-key read of the run and a range of its own `step_runs` rows. This @@ -41,7 +43,7 @@ Each open step gets one kind. These five are causes: | --- | --- | --- | | `ask` | An open `work ask`, or a human gate, waits for an answer. This includes the step that asked. | the person | | `seat-down` | The seat that holds the step, or that would take it, is stopped, failed, suspended, down, retired or faulted. | the seat | -| `no-progress` | A seat holds the step and has recorded no progress (or claim) for longer than the quiet limit. | the holder | +| `no-progress` | A seat holds the step and has recorded no progress (or claim) for longer than the quiet limit, or a program step's journal has no new entry within its `stalled-after`. | the holder (Fractal for a program) | | `unclaimed` | Ready for longer than the quiet limit, and nobody claims it. | the assignee or the queue holding it | | `hold` | Blocked with a reason, in a declared `work wait`, in a retry backoff, or Fractal's wait for another run. | outside the fleet | @@ -50,6 +52,7 @@ The rest are not causes: | Kind | Meaning | | --- | --- | | `working` | Progress within the quiet limit. | +| `program` | A program step whose journal moved within its `stalled-after`. The detail names its innermost open key: the most recently started key without a result, and the child run or ask it waits on. A program's ask is an `ask`, named with its key. | | `verifying` | Submitted, and Fractal checks its gates. | | `ready` | Ready within the quiet limit. | | `queued` | Ready, waiting its turn behind other work on a busy seat. | diff --git a/release-notes/program-step-faults.json b/release-notes/program-step-faults.json new file mode 100644 index 00000000000..1459046395d --- /dev/null +++ b/release-notes/program-step-faults.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "A program step whose journal is quiet past its `stalled-after` raises the `program/stalled` fault naming its open key, and a program step whose host cannot start or whose worker missed its liveness deadline raises `program/host`; both close when the cause ends or the run ends", + "replay": {"status": "none", "detail": "No new claim kind. The faults are ordinary `reconcile.fault` claims on the program's step-run subject with the new scopes `program/stalled` and `program/host`."}, + "schema": {"status": "none", "detail": "No table, index or migration change."}, + "rules": {"status": "none", "detail": "No rule or authority change."}, + "compatibility": {"status": "changed", "detail": "A program step's evaluation arms a deadline at its last journal entry plus `stalled-after` (default 30 minutes); the stalled check runs at that deadline, not on a polling loop. A host start failure (no runtime, a filesystem error or a failed spawn) used to be logged only, with no fault recorded; it is now the `program/host` fault on the step, and while that fault is open the step is dispatched again every 30 seconds until a host starts, the step times out or the run is cancelled. The inactive-fault sweep closes both scopes once the step's run ends. Faults that last surface as operator attention like other reconcile faults."}, + "downtime": {"status": "none", "detail": "No restart or migration beyond a normal deploy."}, + "manual": {"status": "none", "detail": "No operator action is required."}, + "recovery": {"status": "none", "detail": "A fault closes by itself when the journal moves, a host runs the step, or the run ends."} +} diff --git a/release-notes/program-step-visibility.json b/release-notes/program-step-visibility.json new file mode 100644 index 00000000000..bd0119d77cc --- /dev/null +++ b/release-notes/program-step-visibility.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "`fractal missions show` lists a program step's journal keys under the step (status, duration, failure, the child run or ask a key waits on) and `fractal missions waiting-on` names its innermost open key; a program step is never shown as `watching`, and one whose journal has no new entry within its `stalled-after` stalls its run", + "replay": {"status": "none", "detail": "No claim kind or projection change. Views read the existing `program.step` claims through the subject/kind claims page the engine's replay already uses."}, + "schema": {"status": "none", "detail": "No table, index or migration change. The client v0 schema gains the work `gate_kind` value `program` and the optional work field `program`."}, + "rules": {"status": "none", "detail": "No rule or authority change."}, + "compatibility": {"status": "changed", "detail": "`GET /v1/mission-runs/{run}` and the mission tree carry `programs` (omitted when a run has no program step). Published work rows of program steps carry `gate_kind: program` instead of `watch`, and a `program` object (innermost open key, its start, its child, last entry, stalled-after); clients that matched `watch` for agentless steps without gates now see `program` for program steps. Waiting-on adds the open-step kind `program`; a program step past its `stalled-after` (default 30 minutes) is `no-progress`. Generated TypeScript, Swift and Rust clients model the new field."}, + "downtime": {"status": "none", "detail": "No restart or migration beyond a normal deploy."}, + "manual": {"status": "none", "detail": "No operator action is required."}, + "recovery": {"status": "none", "detail": "The fields are read-only presentation; reverting removes them."} +}