Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
10 changes: 8 additions & 2 deletions docs/CONTROLLED_SESSION_DESIGN.md
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
---
status: Active
updated: 2026-08-09
updated: 2026-08-10
summary: Capability-scoped execution sessions that inherit Reploy's global container sandbox.
---

Expand Down Expand Up @@ -55,7 +55,13 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta
completion and result acknowledgement, and removes both containers and the
private channel. A workload that starts before a later startup step fails is
still terminated and its output is finalized through the same barrier.
Crash watchdogs and restart reconciliation remain the next ownership phase;
Before creating any session resource, the planned controller, workload, and
private-channel ownership plus the session, lease, and boot identities are
now durably recorded in the existing live-run state. Reploy monotonically
fills each exact full container ID after Docker creates it, and both IDs are
durable before either process starts. Verified cleanup removes that record;
failed or unverifiable partial-preparation cleanup retains it. The watchdog
and restart reconciliation remain the next ownership phases, and
controlled-session networking remains a later phase.
- Initial runtime: Linux containers under Docker
- Motivating clients: OmegaFlow recording, sandboxed AI agents, security
Expand Down
107 changes: 101 additions & 6 deletions internal/deploy/live_run_queue.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"errors"
"fmt"
"io"
"path/filepath"
"regexp"

"github.com/omry/reploy/internal/canonical"
Expand Down Expand Up @@ -55,9 +56,28 @@ type LiveRunV1 struct {
}

type LiveRunQueueV1 struct {
Schema string `json:"schema"`
Runs []LiveRunV1 `json:"runs"`
Cleanup []LiveRunContainerCleanupV1 `json:"cleanup,omitempty"`
Schema string `json:"schema"`
Runs []LiveRunV1 `json:"runs"`
ControlledSessions []ControlledSessionOwnershipV1 `json:"controlled_sessions,omitempty"`
Cleanup []LiveRunContainerCleanupV1 `json:"cleanup,omitempty"`
}

type ControlledSessionOwnershipV1 struct {
LiveRunID string `json:"live_run_id"`
BootSession string `json:"boot_session"`
SessionHandle string `json:"session_handle"`
ChannelDirectory string `json:"channel_directory"`
Controller ControlledSessionContainerOwnershipV1 `json:"controller"`
Workload ControlledSessionContainerOwnershipV1 `json:"workload"`
}

type ControlledSessionContainerOwnershipV1 struct {
Role string `json:"role"`
ID string `json:"id"`
Name string `json:"name"`
DeploymentID string `json:"deployment_id"`
GenerationReference string `json:"generation_reference"`
BuildIdentity string `json:"build_identity"`
}

type LiveRunRecoveryReasonV1 string
Expand Down Expand Up @@ -98,6 +118,9 @@ var ErrLiveRunConflict = errors.New("another run must finish first")

var liveRunIDPatternV1 = regexp.MustCompile(`^run-[0-9a-f]{16}$`)
var controlMarkerIDPatternV1 = regexp.MustCompile(`^control-[0-9a-f]{16}$`)
var controlledSessionHandlePatternV1 = regexp.MustCompile(`^session-[0-9a-f]{64}$`)
var controlledSessionContainerIDPatternV1 = regexp.MustCompile(`^[0-9a-f]{64}$`)
var controlledSessionBuildIdentityPatternV1 = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)

func NewLiveRunQueueV1() LiveRunQueueV1 {
return LiveRunQueueV1{Schema: LiveRunQueueSchemaV1, Runs: []LiveRunV1{}}
Expand Down Expand Up @@ -189,6 +212,14 @@ func ValidateLiveRunQueueV1(queue LiveRunQueueV1) error {
if queue.Runs == nil {
return fmt.Errorf("live run queue runs must use an array")
}
for index, ownership := range queue.ControlledSessions {
if err := validateControlledSessionOwnershipV1(ownership); err != nil {
return fmt.Errorf("live run queue controlled session %d: %w", index, err)
}
if index > 0 && queue.ControlledSessions[index-1].LiveRunID >= ownership.LiveRunID {
return fmt.Errorf("live run queue controlled sessions must be sorted and unique by live run ID")
}
}
for index, cleanup := range queue.Cleanup {
if err := validateLiveRunContainerCleanupV1(cleanup); err != nil {
return fmt.Errorf("live run queue cleanup entry %d: %w", index, err)
Expand Down Expand Up @@ -247,6 +278,65 @@ func ValidateLiveRunQueueV1(queue LiveRunQueueV1) error {
return nil
}

func validateControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1) error {
if err := ValidateLiveRunIDV1(ownership.LiveRunID); err != nil {
return fmt.Errorf("live run ID: %w", err)
}
if err := validateBootSessionIDV1(ownership.BootSession); err != nil {
return err
}
if !controlledSessionHandlePatternV1.MatchString(ownership.SessionHandle) {
return fmt.Errorf("session handle must use session- followed by 64 lowercase hexadecimal characters")
}
if !filepath.IsAbs(ownership.ChannelDirectory) || filepath.Clean(ownership.ChannelDirectory) != ownership.ChannelDirectory || !safeRecoveryIdentity(ownership.ChannelDirectory) {
return fmt.Errorf("channel directory must be a clean absolute path")
}
if err := validateControlledSessionContainerOwnershipStateV1(ownership.Controller, "controller"); err != nil {
return fmt.Errorf("controller: %w", err)
}
if err := validateControlledSessionContainerOwnershipStateV1(ownership.Workload, "workload"); err != nil {
return fmt.Errorf("workload: %w", err)
}
if ownership.Controller.ID == "" && ownership.Workload.ID != "" {
return fmt.Errorf("workload container ID cannot be recorded before the controller container ID")
}
if ownership.Controller.ID != "" && ownership.Workload.ID != "" && ownership.Controller.ID == ownership.Workload.ID {
return fmt.Errorf("controller and workload must name different containers")
}
return nil
}

func validateControlledSessionContainerOwnershipV1(ownership ControlledSessionContainerOwnershipV1, role string) error {
if err := validateControlledSessionContainerOwnershipStateV1(ownership, role); err != nil {
return err
}
if ownership.ID == "" {
return fmt.Errorf("container ID must use 64 lowercase hexadecimal characters")
}
return nil
}

func validateControlledSessionContainerOwnershipStateV1(ownership ControlledSessionContainerOwnershipV1, role string) error {
if ownership.Role != role {
return fmt.Errorf("role must be %q", role)
}
if ownership.ID != "" && !controlledSessionContainerIDPatternV1.MatchString(ownership.ID) {
return fmt.Errorf("container ID must use 64 lowercase hexadecimal characters")
}
for label, value := range map[string]string{
"name": ownership.Name, "deployment ID": ownership.DeploymentID,
"generation reference": ownership.GenerationReference,
} {
if !safeRecoveryIdentity(value) {
return fmt.Errorf("%s must be nonempty safe text", label)
}
}
if !controlledSessionBuildIdentityPatternV1.MatchString(ownership.BuildIdentity) {
return fmt.Errorf("build identity must be a sha256 digest")
}
return nil
}

func validateLiveRunContainerCleanupV1(cleanup LiveRunContainerCleanupV1) error {
if !safeRecoveryIdentity(cleanup.Container) {
return fmt.Errorf("cleanup container must be nonempty safe text")
Expand Down Expand Up @@ -506,14 +596,19 @@ func ControlMarkersV1(queue LiveRunQueueV1) []ControlMarkerV1 {
}

func cloneLiveRunQueueV1(queue LiveRunQueueV1) LiveRunQueueV1 {
var controlledSessions []ControlledSessionOwnershipV1
if queue.ControlledSessions != nil {
controlledSessions = append([]ControlledSessionOwnershipV1{}, queue.ControlledSessions...)
}
var cleanup []LiveRunContainerCleanupV1
if queue.Cleanup != nil {
cleanup = append([]LiveRunContainerCleanupV1{}, queue.Cleanup...)
}
return LiveRunQueueV1{
Schema: queue.Schema,
Runs: append([]LiveRunV1{}, queue.Runs...),
Cleanup: cleanup,
Schema: queue.Schema,
Runs: append([]LiveRunV1{}, queue.Runs...),
ControlledSessions: controlledSessions,
Cleanup: cleanup,
}
}

Expand Down
151 changes: 150 additions & 1 deletion internal/deploy/live_run_queue_file.go
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,155 @@ func (lock *OperationLock) RecordLiveRunContainerV1(id string, container string)
return fmt.Errorf("live run %q is not outstanding", id)
}

// RecordControlledSessionOwnershipV1 durably binds the planned resources to an
// active admitted shell and monotonically fills each exact container ID after
// Docker returns it. The boot identity comes from the admitted run already
// protected by this lock.
func (lock *OperationLock) RecordControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1) (ControlledSessionOwnershipV1, error) {
if lock == nil {
return ControlledSessionOwnershipV1{}, fmt.Errorf("record controlled session ownership requires an operation lock")
}
if err := ValidateLiveRunIDV1(ownership.LiveRunID); err != nil {
return ControlledSessionOwnershipV1{}, err
}
lock.mutex.Lock()
defer lock.mutex.Unlock()
path, err := lock.liveRunQueuePathLockedV1()
if err != nil {
return ControlledSessionOwnershipV1{}, err
}
queue, _, err := readLiveRunQueuePathV1(path)
if err != nil {
return ControlledSessionOwnershipV1{}, err
}
var admitted *LiveRunV1
for index := range queue.Runs {
if queue.Runs[index].ID == ownership.LiveRunID {
admitted = &queue.Runs[index]
break
}
}
if admitted == nil {
return ControlledSessionOwnershipV1{}, fmt.Errorf("live run %q is not outstanding", ownership.LiveRunID)
}
if admitted.Status != LiveRunStatusActiveV1 || admitted.Kind != LiveRunKindShellV1 {
return ControlledSessionOwnershipV1{}, fmt.Errorf("controlled session live run %q must be an active shell", ownership.LiveRunID)
}
if admitted.Container != "" {
return ControlledSessionOwnershipV1{}, fmt.Errorf("controlled session live run %q already names container %q", ownership.LiveRunID, admitted.Container)
}
Comment thread
omry marked this conversation as resolved.
if admitted.GenerationReference != ownership.Workload.GenerationReference {
return ControlledSessionOwnershipV1{}, fmt.Errorf("controlled session workload generation does not match admitted live run %q", ownership.LiveRunID)
}
ownership.BootSession = admitted.BootSession
if err := validateControlledSessionOwnershipV1(ownership); err != nil {
return ControlledSessionOwnershipV1{}, err
}
insert := sort.Search(len(queue.ControlledSessions), func(index int) bool {
return queue.ControlledSessions[index].LiveRunID >= ownership.LiveRunID
})
if insert < len(queue.ControlledSessions) && queue.ControlledSessions[insert].LiveRunID == ownership.LiveRunID {
merged, err := mergeControlledSessionOwnershipV1(queue.ControlledSessions[insert], ownership)
if err != nil {
return ControlledSessionOwnershipV1{}, fmt.Errorf("live run %q already has different controlled-session ownership: %w", ownership.LiveRunID, err)
}
if merged == queue.ControlledSessions[insert] {
return merged, nil
}
queue.ControlledSessions[insert] = merged
if err := commitLiveRunQueuePathV1(path, queue); err != nil {
return ControlledSessionOwnershipV1{}, err
}
return merged, nil
}
queue.ControlledSessions = append(queue.ControlledSessions, ControlledSessionOwnershipV1{})
copy(queue.ControlledSessions[insert+1:], queue.ControlledSessions[insert:])
queue.ControlledSessions[insert] = ownership
if err := commitLiveRunQueuePathV1(path, queue); err != nil {
return ControlledSessionOwnershipV1{}, err
}
return ownership, nil
}

func mergeControlledSessionOwnershipV1(
existing ControlledSessionOwnershipV1,
requested ControlledSessionOwnershipV1,
) (ControlledSessionOwnershipV1, error) {
existingPlan := existing
requestedPlan := requested
existingPlan.Controller.ID = ""
existingPlan.Workload.ID = ""
requestedPlan.Controller.ID = ""
requestedPlan.Workload.ID = ""
if existingPlan != requestedPlan {
return ControlledSessionOwnershipV1{}, fmt.Errorf("immutable resource plan changed")
}
merged := existing
mergeID := func(current string, next string, role string) (string, error) {
if next == "" {
return current, nil
}
if current != "" && current != next {
return "", fmt.Errorf("%s container ID changed", role)
}
return next, nil
}
var err error
merged.Controller.ID, err = mergeID(existing.Controller.ID, requested.Controller.ID, "controller")
if err != nil {
return ControlledSessionOwnershipV1{}, err
}
merged.Workload.ID, err = mergeID(existing.Workload.ID, requested.Workload.ID, "workload")
if err != nil {
return ControlledSessionOwnershipV1{}, err
}
if err := validateControlledSessionOwnershipV1(merged); err != nil {
return ControlledSessionOwnershipV1{}, err
}
return merged, nil
}

// CompleteControlledSessionV1 atomically removes a verified-clean session's
// ownership record and admitted run. Failed cleanup must not call this method.
func (lock *OperationLock) CompleteControlledSessionV1(id string) (bool, error) {
if lock == nil {
return false, fmt.Errorf("complete controlled session requires an operation lock")
}
if err := ValidateLiveRunIDV1(id); err != nil {
return false, err
}
lock.mutex.Lock()
defer lock.mutex.Unlock()
path, err := lock.liveRunQueuePathLockedV1()
if err != nil {
return false, err
}
queue, _, err := readLiveRunQueuePathV1(path)
if err != nil {
return false, err
}
updated, runRemoved, err := RemoveLiveRunV1(queue, id)
if err != nil {
return false, err
}
ownershipRemoved := false
for index, ownership := range updated.ControlledSessions {
if ownership.LiveRunID != id {
continue
}
updated.ControlledSessions = append(updated.ControlledSessions[:index], updated.ControlledSessions[index+1:]...)
ownershipRemoved = true
break
}
if !runRemoved && !ownershipRemoved {
return false, nil
}
if err := commitLiveRunQueuePathV1(path, updated); err != nil {
return false, err
}
return true, nil
}

func (lock *OperationLock) RemoveLiveRunV1(id string) (LiveRunQueueV1, bool, error) {
if lock == nil {
return LiveRunQueueV1{}, false, fmt.Errorf("remove live run requires an operation lock")
Expand Down Expand Up @@ -518,7 +667,7 @@ func commitLiveRunQueuePathV1(path string, queue LiveRunQueueV1) error {
if err != nil {
return err
}
if len(queue.Runs) == 0 && len(queue.Cleanup) == 0 {
if len(queue.Runs) == 0 && len(queue.ControlledSessions) == 0 && len(queue.Cleanup) == 0 {
return removeLiveRunQueuePathV1(path)
}
if err := writeAtomicStateFile(path, content, 0o600); err != nil {
Expand Down
Loading
Loading