Skip to content

Enforce controlled-session output finalization #23

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
omry wants to merge 1 commit into from
+370 −73
Open

Enforce controlled-session output finalization #23

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions docs/CONTROLLED_SESSION_DESIGN.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,9 +10,10 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta

- Decision state: Focused review complete; high-level decisions approved
- Implementation state: Initial global sandbox prerequisites, trusted
application-startup verification, controlled-session authorization, and the
initial framed protocol are implemented; lifecycle, controlled networking,
and Docker orchestration remain later slices
application-startup verification, controlled-session authorization, the
framed protocol, and lifecycle state machine, including its output-finalization
barrier and timeout outcome, are implemented; bounded output draining,
controlled-session networking, and Docker orchestration remain later slices
Comment on lines 12 to +16
- Initial runtime: Linux containers under Docker
- Motivating clients: OmegaFlow recording, sandboxed AI agents, security
inspection, and untrusted-code execution
Expand Down Expand Up @@ -448,8 +449,8 @@ cleanup and no canceled request is replayed.

### Session Events

- `opened`: reports the effective dimensions, identity, generation, and fixed
session capabilities.
- `opened`: reports the effective dimensions, identity, generation, fixed
session capabilities, and workload-output-finalization timeout.
- `output(bytes)`: ordered PTY output bytes.
- `workload_exit(status, reason)`: reports host-observed workload-shell
exit.
Expand Down Expand Up @@ -502,10 +503,9 @@ Host Reploy owns workload-output finalization; it never waits indefinitely for
workload cooperation. Once termination begins, it rejects new output surfaces,
performs bounded graceful shutdown followed by forced container stop, and
continues draining the PTY. The immutable session plan carries a finite
output-finalization deadline. The initial implementation may use a fixed
host-owned value, but the effective value is reported by `opened` and applies
to workload shutdown, final buffered-byte delivery, and controller
backpressure.
output-finalization deadline. Protocol v1 defines an initial host-owned default
of 30 seconds; the effective value is reported by `opened` and applies to
workload shutdown, final buffered-byte delivery, and controller backpressure.

If every final byte is delivered and every output surface reaches EOF before
the deadline, Host Reploy emits `workload_outputs_finalized(drained)` only after
Expand Down
151 changes: 98 additions & 53 deletions internal/controlledsession/lifecycle.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,16 +26,17 @@ type FinishV1 struct {
type ObservationKindV1 string

const (
ObservationActivatedV1 ObservationKindV1 = "activated"
ObservationWorkloadExitV1 ObservationKindV1 = "workload-exit"
ObservationHostCancelV1 ObservationKindV1 = "host-cancel"
ObservationControllerLostV1 ObservationKindV1 = "controller-lost"
ObservationRuntimeObservationLostV1 ObservationKindV1 = "runtime-observation-lost"
ObservationStartupFailureV1 ObservationKindV1 = "startup-failure"
ObservationWorkloadOutputsFinalizedV1 ObservationKindV1 = "workload-outputs-finalized"
ObservationControllerFinalizationExpiredV1 ObservationKindV1 = "controller-finalization-expired"
ObservationFinishedV1 ObservationKindV1 = "finished"
ObservationResultDeliveredV1 ObservationKindV1 = "result-delivered"
ObservationActivatedV1 ObservationKindV1 = "activated"
ObservationWorkloadExitV1 ObservationKindV1 = "workload-exit"
ObservationHostCancelV1 ObservationKindV1 = "host-cancel"
ObservationControllerLostV1 ObservationKindV1 = "controller-lost"
ObservationRuntimeObservationLostV1 ObservationKindV1 = "runtime-observation-lost"
ObservationStartupFailureV1 ObservationKindV1 = "startup-failure"
ObservationWorkloadOutputsFinalizedV1 ObservationKindV1 = "workload-outputs-finalized"
ObservationWorkloadOutputFinalizationExpiredV1 ObservationKindV1 = "workload-output-finalization-expired"
ObservationControllerFinalizationExpiredV1 ObservationKindV1 = "controller-finalization-expired"
ObservationFinishedV1 ObservationKindV1 = "finished"
ObservationResultDeliveredV1 ObservationKindV1 = "result-delivered"
)

type ObservationV1 struct {
Expand All @@ -47,30 +48,32 @@ type ObservationV1 struct {
}

type SnapshotV1 struct {
State StateV1
Cause TerminationCauseV1
WorkloadStatus ProcessStatusV1
WorkloadOutputFinalizationStatus WorkloadOutputFinalizationStatusV1
RuntimeObservationStatus RuntimeObservationStatusV1
ControllerFinalizationStatus ControllerFinalizationStatusV1
AwaitingControllerFinalization bool
AwaitingResultAcknowledgement bool
ResultAcknowledged bool
Result *ResultV1
State StateV1
Cause TerminationCauseV1
WorkloadStatus ProcessStatusV1
WorkloadOutputFinalizationStatus WorkloadOutputFinalizationStatusV1
RuntimeObservationStatus RuntimeObservationStatusV1
ControllerFinalizationStatus ControllerFinalizationStatusV1
AwaitingWorkloadOutputFinalization bool
AwaitingControllerFinalization bool
AwaitingResultAcknowledgement bool
ResultAcknowledged bool
Result *ResultV1
}

type TransitionV1 struct {
Before StateV1
After StateV1
Cause TerminationCauseV1
CauseLatched bool
BeginTermination bool
WorkloadOutputFinalizationStatus WorkloadOutputFinalizationStatusV1
AwaitingControllerFinalization bool
AwaitingResultAcknowledgement bool
RequestAccepted bool
ResultAcknowledged bool
Result *ResultV1
Before StateV1
After StateV1
Cause TerminationCauseV1
CauseLatched bool
BeginTermination bool
WorkloadOutputFinalizationStatus WorkloadOutputFinalizationStatusV1
AwaitingWorkloadOutputFinalization bool
AwaitingControllerFinalization bool
AwaitingResultAcknowledgement bool
RequestAccepted bool
ResultAcknowledged bool
Result *ResultV1
}

var ErrRequestRejected = errors.New("controlled-session request rejected")
Expand All @@ -89,6 +92,7 @@ type MachineV1 struct {
workloadOutputs WorkloadOutputFinalizationStatusV1
controller ControllerFinalizationStatusV1
runtimeObservation RuntimeObservationStatusV1
waitingOutputs bool
waitingFinalize bool
resultDelivered bool
resultAcknowledged bool
Expand Down Expand Up @@ -145,7 +149,10 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro
if observation.WorkloadStatus == nil || observation.WorkloadOutputFinalizationStatus != nil || observation.Reason != "" || observation.Finish != nil {
return transition, fmt.Errorf("%w: workload exit requires exactly one workload status", ErrObservationRejected)
}
if machine.workloadOutputs.Kind != "" {
if machine.workloadOutputs.Kind != "" && !(machine.activated &&
machine.state == StateTerminatingV1 &&
machine.workload.Kind == ProcessStatusUnknownV1 &&
machine.workloadOutputs.Kind == WorkloadOutputFinalizationFailedV1) {
Comment on lines +152 to +155

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Accept workload exit after output-finalization expiry

When termination starts from a terminate, host-cancel, or controller-loss action and the output deadline expires before the forced workload stop is observed, ObservationWorkloadOutputFinalizationExpiredV1 records failed output while the workload status remains unknown. This guard then rejects the subsequent ObservationWorkloadExitV1 because its only exception is runtime-observation loss. Since the deadline explicitly covers workload shutdown, this is a valid ordering and prevents the authoritative workload-exit event/status from being recorded after a timeout; allow a late terminal workload observation for an activated session whose output finalization failed.

Useful? React with 👍 / 👎.

return transition, fmt.Errorf("%w: workload exit cannot follow workload output finalization", ErrObservationRejected)
}
if machine.state == StatePreparingV1 {
Expand Down Expand Up @@ -186,13 +193,14 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro
machine.runtimeObservation = RuntimeObservationStatusV1{Kind: RuntimeObservationLostV1, Reason: observation.Reason}
}
machine.latchLocked(CauseRuntimeObservationLostV1, &transition)
// Observation loss does not close an activated workload's output
// surfaces. Keep that barrier pending until the supervisor explicitly
// reports failed closure or its bounded finalization deadline expires.
machine.finalizePreActivationOutputsForRuntimeObservationLossLocked(observation.Reason)
case ObservationStartupFailureV1:
if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus != nil || observation.Finish != nil {
return transition, fmt.Errorf("%w: startup failure carries only a reason", ErrObservationRejected)
}
if machine.workloadOutputs.Kind != "" {
return transition, fmt.Errorf("%w: startup failure cannot follow workload output finalization", ErrObservationRejected)
}
if err := validateRequiredSafeTextV1("startup-failure reason", observation.Reason); err != nil {
return transition, fmt.Errorf("%w: %v", ErrObservationRejected, err)
}
Expand All @@ -202,16 +210,12 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro
machine.controller = ControllerFinalizationStatusV1{Kind: ControllerFinalizationStartupFailedV1, Reason: observation.Reason}
machine.latchLocked(CauseStartupFailureV1, &transition)
case ObservationWorkloadOutputsFinalizedV1:
if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus == nil || observation.Reason != "" || observation.Finish != nil {
return transition, fmt.Errorf("%w: workload output finalization requires exactly one status", ErrObservationRejected)
if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus == nil || observation.Reason != "" || observation.Finish != nil || !machine.waitingOutputs {
return transition, fmt.Errorf("%w: workload output finalization requires exactly one status while output finalization is pending", ErrObservationRejected)
}
if machine.state != StateTerminatingV1 {
return transition, fmt.Errorf("%w: workload output finalization is valid only while terminating", ErrObservationRejected)
}
if machine.workloadOutputs.Kind != "" {
return transition, fmt.Errorf("%w: workload output finalization was already observed", ErrObservationRejected)
}
if machine.activated && machine.workload.Kind == ProcessStatusUnknownV1 {
if machine.activated && machine.workload.Kind == ProcessStatusUnknownV1 &&
!(machine.runtimeObservation.Kind == RuntimeObservationLostV1 &&
observation.WorkloadOutputFinalizationStatus.Kind == WorkloadOutputFinalizationFailedV1) {
return transition, fmt.Errorf("%w: workload output finalization requires an observed terminal workload status", ErrObservationRejected)
}
if err := validateWorkloadOutputFinalizationStatusV1(*observation.WorkloadOutputFinalizationStatus); err != nil {
Expand All @@ -220,9 +224,18 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro
if machine.runtimeObservation.Kind == RuntimeObservationLostV1 && observation.WorkloadOutputFinalizationStatus.Kind == WorkloadOutputFinalizationDrainedV1 {
return transition, fmt.Errorf("%w: runtime observation loss requires failed workload output finalization", ErrObservationRejected)
}
machine.workloadOutputs = *observation.WorkloadOutputFinalizationStatus
machine.waitingFinalize = machine.controller.Kind == ControllerFinalizationActiveV1 &&
containsOperationV1(machine.authorization.Operations, OperationCompleteV1)
machine.completeOutputFinalizationLocked(*observation.WorkloadOutputFinalizationStatus)
case ObservationWorkloadOutputFinalizationExpiredV1:
if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus != nil || observation.Finish != nil || !machine.waitingOutputs {
return transition, fmt.Errorf("%w: output-finalization expiry carries only a required reason while output finalization is pending", ErrObservationRejected)
}
if err := validateRequiredSafeTextV1("workload output finalization expiry reason", observation.Reason); err != nil {
return transition, fmt.Errorf("%w: %v", ErrObservationRejected, err)
}
machine.completeOutputFinalizationLocked(WorkloadOutputFinalizationStatusV1{
Kind: WorkloadOutputFinalizationFailedV1,
Reason: observation.Reason,
})
Comment on lines +228 to +238
case ObservationControllerFinalizationExpiredV1:
if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus != nil || observation.Finish != nil || observation.Reason != "" || !machine.waitingFinalize {
return transition, fmt.Errorf("%w: finalization expiry requires an active controller-finalization wait", ErrObservationRejected)
Expand All @@ -233,8 +246,8 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro
if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus != nil || observation.Reason != "" || observation.Finish == nil {
return transition, fmt.Errorf("%w: finish requires exactly one terminal status set", ErrObservationRejected)
}
if machine.state != StateTerminatingV1 || machine.workloadOutputs.Kind == "" || machine.waitingFinalize {
return transition, fmt.Errorf("%w: finish requires finalized workload output and no pending controller finalization", ErrObservationRejected)
if machine.state != StateTerminatingV1 || machine.waitingOutputs || machine.workloadOutputs.Kind == "" || machine.waitingFinalize {
return transition, fmt.Errorf("%w: finish requires finalized workload output and no pending output or controller finalization", ErrObservationRejected)
}
if err := validateFinishV1(*observation.Finish); err != nil {
return transition, fmt.Errorf("%w: %v", ErrObservationRejected, err)
Expand Down Expand Up @@ -269,6 +282,7 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro
transition.After = machine.state
transition.Cause = machine.cause
transition.WorkloadOutputFinalizationStatus = machine.workloadOutputs
transition.AwaitingWorkloadOutputFinalization = machine.waitingOutputs
transition.AwaitingControllerFinalization = machine.waitingFinalize
transition.AwaitingResultAcknowledgement = machine.resultDelivered && !machine.resultAcknowledged
transition.ResultAcknowledged = machine.resultAcknowledged
Expand Down Expand Up @@ -332,8 +346,8 @@ func (machine *MachineV1) ApplyRequest(request RequestV1) (TransitionV1, error)
// Repeated graceful termination is idempotent and does not alter
// output or controller finalization already in progress.
case RequestCompleteV1:
if !machine.waitingFinalize || machine.controller.Kind != ControllerFinalizationActiveV1 {
return transition, fmt.Errorf("%w: completion is not awaiting controller finalization", ErrRequestRejected)
if machine.waitingOutputs || !machine.waitingFinalize || machine.controller.Kind != ControllerFinalizationActiveV1 {
return transition, fmt.Errorf("%w: completion requires finalized workload outputs and a pending controller finalization", ErrRequestRejected)
}
machine.waitingFinalize = false
machine.controller = ControllerFinalizationStatusV1{Kind: ControllerFinalizationCompletedV1}
Expand All @@ -349,6 +363,7 @@ func (machine *MachineV1) ApplyRequest(request RequestV1) (TransitionV1, error)
transition.After = machine.state
transition.Cause = machine.cause
transition.WorkloadOutputFinalizationStatus = machine.workloadOutputs
transition.AwaitingWorkloadOutputFinalization = machine.waitingOutputs
transition.AwaitingControllerFinalization = machine.waitingFinalize
transition.AwaitingResultAcknowledgement = machine.resultDelivered && !machine.resultAcknowledged
transition.RequestAccepted = true
Expand All @@ -362,6 +377,13 @@ func (machine *MachineV1) latchLocked(cause TerminationCauseV1, transition *Tran
}
machine.cause = cause
machine.state = StateTerminatingV1
if transition.Before == StateActiveV1 {
machine.waitingOutputs = true
} else {
// No workload ran, so there are no workload-originated surfaces to
// drain. Record the barrier as satisfied rather than inventing a wait.
machine.workloadOutputs = WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationDrainedV1}
}
transition.CauseLatched = true
transition.BeginTermination = true
}
Expand All @@ -372,8 +394,9 @@ func (machine *MachineV1) snapshotLocked() SnapshotV1 {
WorkloadOutputFinalizationStatus: machine.workloadOutputs,
RuntimeObservationStatus: machine.runtimeObservation,
ControllerFinalizationStatus: machine.controller, AwaitingControllerFinalization: machine.waitingFinalize,
AwaitingResultAcknowledgement: machine.resultDelivered && !machine.resultAcknowledged,
ResultAcknowledged: machine.resultAcknowledged, Result: cloneResultV1(machine.result),
AwaitingWorkloadOutputFinalization: machine.waitingOutputs,
AwaitingResultAcknowledgement: machine.resultDelivered && !machine.resultAcknowledged,
ResultAcknowledged: machine.resultAcknowledged, Result: cloneResultV1(machine.result),
}
}

Expand All @@ -400,6 +423,28 @@ func validateFinishV1(finish FinishV1) error {
return validateCleanupResultV1(finish.CleanupStatus, finish.RecoveryAction)
}

func (machine *MachineV1) finalizePreActivationOutputsForRuntimeObservationLossLocked(reason string) {
if machine.activated ||
machine.cause != CauseRuntimeObservationLostV1 ||
machine.workloadOutputs.Kind != WorkloadOutputFinalizationDrainedV1 {
return
}
if reason == "" {
reason = "runtime observation was lost before workload output finalization completed"
}
machine.completeOutputFinalizationLocked(WorkloadOutputFinalizationStatusV1{
Kind: WorkloadOutputFinalizationFailedV1,
Reason: reason,
})
}

func (machine *MachineV1) completeOutputFinalizationLocked(status WorkloadOutputFinalizationStatusV1) {
machine.workloadOutputs = status
machine.waitingOutputs = false
machine.waitingFinalize = machine.controller.Kind == ControllerFinalizationActiveV1 &&
containsOperationV1(machine.authorization.Operations, OperationCompleteV1)
}

func equalProcessStatusV1(left ProcessStatusV1, right ProcessStatusV1) bool {
if left.Kind != right.Kind || left.Reason != right.Reason || (left.Code == nil) != (right.Code == nil) {
return false
Expand Down
Loading
Loading