Skip to content
Merged
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
1 change: 1 addition & 0 deletions apps/api/src/audit/enums/audit-action.enum.ts
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ export enum AuditAction {
CREATE_SSH_ACCESS = 'create_ssh_access',
REVOKE_SSH_ACCESS = 'revoke_ssh_access',
RECOVER = 'recover',
FORCE_STOP = 'force_stop',
REGENERATE_PROXY_API_KEY = 'regenerate_proxy_api_key',
REGENERATE_SSH_GATEWAY_API_KEY = 'regenerate_ssh_gateway_api_key',
REGENERATE_SNAPSHOT_MANAGER_CREDENTIALS = 'regenerate_snapshot_manager_credentials',
Expand Down
20 changes: 19 additions & 1 deletion apps/api/src/sandbox/controllers/sandbox.controller.auth.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,21 @@ describe('[AUTH] SandboxController', () => {
])
})

it('forceStopSandbox', () => {
const methodName = trackMethod('forceStopSandbox')
expect(isPublicEndpoint(SandboxController, methodName)).toBe(false)
expectArrayMatch(getAllowedAuthStrategies(SandboxController, methodName), [
AuthStrategyType.API_KEY,
AuthStrategyType.JWT,
])
expectArrayMatch(getAuthContextGuards(SandboxController, methodName), [OrganizationAuthContextGuard])
expectArrayMatch(getResourceAccessGuards(SandboxController, methodName), [SandboxAccessGuard])
expect(getRequiredOrganizationMemberRole(SandboxController, methodName)).toBeUndefined()
expectArrayMatch(getRequiredOrganizationResourcePermissions(SandboxController, methodName), [
OrganizationResourcePermission.WRITE_SANDBOXES,
])
})

it('startSandbox', () => {
const methodName = trackMethod('startSandbox')
expect(isPublicEndpoint(SandboxController, methodName)).toBe(false)
Expand Down Expand Up @@ -421,7 +436,10 @@ describe('[AUTH] SandboxController', () => {
const methodName = trackMethod('validateSshAccess')
expect(isPublicEndpoint(SandboxController, methodName)).toBe(false)
expectArrayMatch(getAllowedAuthStrategies(SandboxController, methodName), [AuthStrategyType.API_KEY])
expectArrayMatch(getAuthContextGuards(SandboxController, methodName), [SshGatewayAuthContextGuard])
expectArrayMatch(getAuthContextGuards(SandboxController, methodName), [
SshGatewayAuthContextGuard,
ProxyAuthContextGuard,
])
})

it('getToolboxProxyUrl', () => {
Expand Down
38 changes: 37 additions & 1 deletion apps/api/src/sandbox/controllers/sandbox.controller.ts
Original file line number Diff line number Diff line change
Expand Up @@ -480,6 +480,42 @@ export class SandboxController {
return sandboxDto
}

@Post(':sandboxIdOrName/force-stop')
@HttpCode(200)
@SkipThrottle({ authenticated: true })
@ThrottlerScope('sandbox-lifecycle')
@ApiOperation({
summary: 'Force-stop a sandbox stuck in a state change',
description:
'Releases the state-change lock of a sandbox stuck in a transient state (e.g. creating, snapshotting) after a runner failure. The sandbox is moved to an error state and can then be recovered or deleted.',
operationId: 'forceStopSandbox',
})
@ApiParam({
name: 'sandboxIdOrName',
description: 'ID or name of the sandbox',
type: 'string',
})
@ApiResponse({
status: 200,
description: 'Sandbox has been force-stopped and moved to error state',
type: SandboxDto,
})
@UseGuards(OrganizationAuthContextGuard, SandboxAccessGuard)
@RequiredOrganizationResourcePermissions([OrganizationResourcePermission.WRITE_SANDBOXES])
@Audit({
action: AuditAction.FORCE_STOP,
targetType: AuditTarget.SANDBOX,
targetIdFromRequest: (req) => req.params.sandboxIdOrName,
targetIdFromResult: (result: SandboxDto) => result?.id,
})
async forceStopSandbox(
@IsOrganizationAuthContext() authContext: OrganizationAuthContext,
@Param('sandboxIdOrName') sandboxIdOrName: string,
): Promise<SandboxDto> {
const sandbox = await this.sandboxService.forceStop(sandboxIdOrName, authContext.organizationId)
return this.sandboxService.toSandboxDto(sandbox)
}

@Post(':sandboxIdOrName/start')
@HttpCode(200)
@SkipThrottle({ authenticated: true })
Expand Down Expand Up @@ -1449,7 +1485,7 @@ export class SandboxController {
type: SshAccessValidationDto,
})
@AuthStrategy(AuthStrategyType.API_KEY)
@UseGuards(SshGatewayAuthContextGuard)
@UseGuards(OrGuard([SshGatewayAuthContextGuard, ProxyAuthContextGuard]))
async validateSshAccess(@Query('token') token: string): Promise<SshAccessValidationDto> {
const result = await this.sandboxService.validateSshAccess(token)
return SshAccessValidationDto.fromValidationResult(result.valid, result.sandboxId)
Expand Down
70 changes: 70 additions & 0 deletions apps/api/src/sandbox/services/sandbox.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,9 @@ import {
} from '../../common/constants/error-messages'
import { RedisLockProvider } from '../common/redis-lock.provider'
import { getStateChangeLockKey } from '../utils/lock-key.util'
import { Job } from '../entities/job.entity'
import { JobStatus } from '../enums/job-status.enum'
import { ResourceType } from '../enums/resource-type.enum'
import { customAlphabet as customNanoid, nanoid, urlAlphabet } from 'nanoid'
import { WithInstrumentation } from '../../common/decorators/otel.decorator'
import { validateMountPaths, validateSubpaths } from '../utils/volume-mount-path-validation.util'
Expand Down Expand Up @@ -153,6 +156,8 @@ export class SandboxService {
private readonly dockerRegistryService: DockerRegistryService,
@InjectRepository(SandboxFork)
private readonly sandboxForkRepository: Repository<SandboxFork>,
@InjectRepository(Job)
private readonly jobRepository: Repository<Job>,
@Inject(SANDBOX_SEARCH_ADAPTER)
private readonly sandboxSearchAdapter: SandboxSearchAdapter,
) {}
Expand Down Expand Up @@ -2187,6 +2192,71 @@ export class SandboxService {
return updatedSandbox
}

/**
* Force-stops a sandbox that is stuck in a transient state (e.g. `creating`,
* `snapshotting`) because a runner crashed or a job can never complete.
*
* Unlike `stop`, this bypasses the state/pending checks — that is its entire
* purpose: it releases the state-change lock by marking the sandbox ERROR
* (which forces `pending=false` via entity invariants) with a clear reason,
* fails any incomplete jobs for the sandbox so the unique incomplete-job
* index no longer blocks new work, and clears the Redis state-change lock.
* The sandbox can then be recovered (`/recover`) or destroyed normally.
*/
async forceStop(sandboxIdOrName: string, organizationId?: string): Promise<Sandbox> {
const sandbox = await this.findOneByIdOrName(sandboxIdOrName, organizationId)

if (String(sandbox.state) === String(sandbox.desiredState) && !sandbox.pending) {
throw new BadRequestError('Sandbox is not stuck in a state change — use the regular stop endpoint instead')
}

if (sandbox.state === SandboxState.DESTROYED || sandbox.desiredState === SandboxDesiredState.DESTROYED) {
throw new BadRequestError('Sandbox is being destroyed and cannot be force-stopped')
}

const lockKey = getStateChangeLockKey(sandbox.id)
if (!(await this.redisLockProvider.lock(lockKey, 60))) {
throw new StateChangeInProgressError()
}

try {
const updatedSandbox = await this.sandboxRepository.updateWhere(sandbox.id, {
updateData: {
state: SandboxState.ERROR,
desiredState: SandboxDesiredState.STOPPED,
errorReason: 'Force-stopped by user while stuck in a state change',
},
whereCondition: { state: sandbox.state },
})

// Fail incomplete jobs so the unique incomplete-job index no longer
// blocks new jobs for this sandbox. Written directly (not via
// JobService.updateJobStatus) so the job-state handler does not run:
// its completion handlers could legitimately move the sandbox to a
// different end state (e.g. SNAPSHOT_SANDBOX restores the pre-snapshot
// state) while force-stop's contract is to land on ERROR + STOPPED.
const failResult = await this.jobRepository.update(
{
resourceType: ResourceType.SANDBOX,
resourceId: sandbox.id,
completedAt: null,
},
{
status: JobStatus.FAILED,
errorMessage: 'Job superseded by force-stop',
completedAt: new Date(),
},
)
if (failResult.affected) {
this.logger.warn(`Force-stop of sandbox ${sandbox.id} failed ${failResult.affected} incomplete job(s)`)
}

Comment on lines +2222 to +2253

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
fd -g '*job*.service.ts' apps/api/src/job apps/api/src/sandbox 2>/dev/null
rg -n "updateJobStatus" apps/api/src -C6

Repository: arrrrny/daytona

Length of output: 8212


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "## job.service updateJobStatus outline"
ast-grep outline apps/api/src/sandbox/services/job.service.ts --match updateJobStatus --view expanded || true

echo "## job.service updateJobStatus implementation"
sed -n '236,380p' apps/api/src/sandbox/services/job.service.ts

echo "## sandbox forceStop and related lines"
sed -n '2180,2280p' apps/api/src/sandbox/services/sandbox.service.ts

echo "## references to runV0SnapshotFromSandbox / persistSnapshotFromSandbox"
rg -n "runV0SnapshotFromSandbox|persistSnapshotFromSandbox|create.*Sandbox|create.*Snapshot|SnapshotService" apps/api/src/sandbox -C5

Repository: arrrrny/daytona

Length of output: 50372


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "## JobStatus isValid transition implementation"
sed -n '350,400p' apps/api/src/sandbox/services/job.service.ts

echo "## job-state-handler outline and relevant completion handling"
ast-grep outline apps/api/src/sandbox/services/job-state-handler.service.ts --view expanded | sed -n '1,220p'
rg -n "handleJobCompletion|SNAPSHOT_SANDBOX|restore|snapshot|SandboxState|SandboxDesiredState" apps/api/src/sandbox/services/job-state-handler.service.ts -C4

echo "## v0 snapshot create implementation"
sed -n '280,360p' apps/api/src/sandbox/runner-adapter/runnerAdapter.v0.ts
rg -n "runV0SnapshotFromSandbox|runV0Snapshot|createSnapshotFromSandbox" apps/api/src/sandbox -C6

echo "## snapshot persistence deps and implementation"
sed -n '1,140p' apps/api/src/sandbox/utils/persist-snapshot-from-sandbox.util.ts
sed -n '1,220p' apps/api/src/sandbox/services/snapshot.service.ts
rg -n "forceStop|snapshot.*sandbox|SandboxState.ERROR|SandboxState.*SNAPSHOT|SNAPSHOT_SANDBOX" apps/api/src/sandbox/services/*.ts apps/api/src/sandbox/managers/*.ts -C4

Repository: arrrrny/daytona

Length of output: 50372


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "## complete forceStop implementation"
rg -n "async forceStop|const validRunners|isEphemeral|createSnapshotManager|pendingSnapshotCountIncrement|runV0SnapshotFromSandbox|force-stop|STORING_SNAPSHOT|SNAPSHOT_SANDBOX|validRunners.forEach|newPromise|pending=false" apps/api/src/sandbox/services/sandbox.service.ts -C5

echo "## runV0SnapshotFromSandbox implementation"
sed -n '1446,1535p' apps/api/src/sandbox/services/sandbox.service.ts

echo "## read-only behavioral simulation of relevant state machines and update path"
python3 - <<'PY'
from enum import Enum
class JobStatus(Enum):
    PENDING = 1
    IN_PROGRESS = 2
    COMPLETED = 3
    FAILED = 4
class SandboxState(Enum):
    SNAPSHOTTING = 1
    ERROR = 2
class SnapshotState(Enum):
    ACTIVE = 3
def valid(current, new):
    return current == new or (current == JobStatus.PENDING and new in (JobStatus.IN_PROGRESS, JobStatus.FAILED)) or (current == JobStatus.IN_PROGRESS and new in (JobStatus.COMPLETED, JobStatus.FAILED))

job_status = JobStatus.IN_PROGRESS
job_status = JobStatus.FAILED
completed_at = "2026-07-01T12:00:00Z"
print("direct failed update result:", {"status": job_status, "completed_at": completed_at})
print("if later normal callback requests FAILED status:", valid(job_status, JobStatus.FAILED))
print("if later normal callback requests COMPLETED status:", valid(job_status, JobStatus.COMPLETED))
print("v0 snapshot persistence condition in simulation (no db state passed):", True)
PY

Repository: arrrrny/daytona

Length of output: 17781


Force-stop should prevent late in-flight snapshots from winning.

JobStatus rejects IN_PROGRESS → COMPLETED after forceStop has set jobs to FAILED, so a delayed job status update cannot move the sandbox back to SNAPSHOTTING. The remaining race is v0 Docker snapshots: runV0SnapshotFromSandbox persists the snapshot and returns previousState: SNAPSHOTTING without checking whether forceStop already set ERROR/STOPPED, so a successful snapshot can still be created after force-stop. Add a shared cancellation signal/check in createSnapshotFromSandbox/v0 snapshot path or gate the final persistence by the sandbox state.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/api/src/sandbox/services/sandbox.service.ts` around lines 2222 - 2253,
Update createSnapshotFromSandbox and the runV0SnapshotFromSandbox path to detect
force-stop cancellation before final snapshot persistence and result
propagation. Recheck the sandbox state immediately before persisting or
otherwise use a shared cancellation signal, and abort the snapshot when it is
already ERROR with desiredState STOPPED so a late snapshot cannot restore
SNAPSHOTTING or win after force-stop.

return updatedSandbox
} finally {
await this.redisLockProvider.unlock(lockKey)
}
}

async pause(sandboxIdOrName: string, organization: Organization): Promise<Sandbox> {
const sandbox = await this.findOneByIdOrName(sandboxIdOrName, organization.id)

Expand Down
1 change: 1 addition & 0 deletions apps/daemon/go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ require (
github.com/kelseyhightower/envconfig v1.4.0
github.com/lmittmann/tint v1.1.2
github.com/mattn/go-isatty v0.0.20
github.com/modelcontextprotocol/go-sdk v1.6.1
github.com/orcaman/concurrent-map/v2 v2.0.1
github.com/pkg/sftp v1.13.6
github.com/ramr/go-reaper v0.3.1
Expand Down
3 changes: 3 additions & 0 deletions apps/daemon/go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,8 @@ github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27k
github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/modelcontextprotocol/go-sdk v1.6.1 h1:0zOSupjKUxPKSocPT1Wtago+mUHU2/uZ4xSOY0FGReU=
github.com/modelcontextprotocol/go-sdk v1.6.1/go.mod h1:kzm3kzFL1/+AziGOE0nUs3gvPoNxMCvkxokMkuFapXQ=
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
Expand Down Expand Up @@ -321,6 +323,7 @@ golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGm
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c=
golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
Expand Down
163 changes: 163 additions & 0 deletions apps/daemon/pkg/session/exec_support.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,163 @@
// Copyright 2025 Daytona Platforms Inc.
// SPDX-License-Identifier: AGPL-3.0

package session

import (
"errors"
"fmt"
"os"
"path/filepath"
"strconv"
"strings"
"syscall"

common_errors "github.com/daytonaio/common-go/pkg/errors"
)

// inputHolderPidFileName is written by cmdWrapperFormat next to the command's
// log file and holds the PID of the async stdin-holder process.
const inputHolderPidFileName = "input_holder.pid"

// CommandLogPaths returns the log file and exit-code file paths for a
// command, so streaming consumers (e.g. the exec WebSocket endpoint) can tail
// output and detect completion without going through the REST endpoints.
func (s *SessionService) CommandLogPaths(sessionId, commandId string) (logPath, exitCodePath string, err error) {
session, ok := s.sessions.Get(sessionId)
if !ok {
return "", "", common_errors.NewNotFoundError(errors.New("session not found"))
}

command, ok := session.commands.Get(commandId)
if !ok {
return "", "", common_errors.NewNotFoundError(errors.New("command not found"))
}

logPath, exitCodePath = command.LogFilePath(session.Dir(s.configDir))
return logPath, exitCodePath, nil
}

// WriteInput writes raw bytes to a running command's stdin FIFO. Unlike
// SendInput it adds no trailing newline and does not echo into the log —
// semantics required by byte-exact protocols (exec-over-WebSocket stdin
// frames). The FIFO is opened non-blocking first so a missing reader
// (command already gone) fails fast instead of hanging the caller.
func (s *SessionService) WriteInput(sessionId, commandId string, data []byte) error {
session, ok := s.sessions.Get(sessionId)
if !ok {
return common_errors.NewNotFoundError(errors.New("session not found"))
}

if session.cmd == nil || session.cmd.Process == nil {
return common_errors.NewGoneError(errors.New("session process is not running"))
}

if session.cmd.ProcessState != nil && session.cmd.ProcessState.Exited() {
return common_errors.NewGoneError(errors.New("session process has exited"))
}

command, ok := session.commands.Get(commandId)
if !ok {
return common_errors.NewNotFoundError(errors.New("command not found"))
}

if command.ExitCode != nil {
return common_errors.NewGoneError(fmt.Errorf("command has already completed with exit code %d", *command.ExitCode))
}

inputFilePath := command.InputFilePath(session.Dir(s.configDir))

fd, err := syscall.Open(inputFilePath, syscall.O_WRONLY|syscall.O_NONBLOCK, 0)
if err != nil {
if errors.Is(err, syscall.ENXIO) || os.IsNotExist(err) {
return common_errors.NewGoneError(errors.New("command stdin is closed"))
}
return common_errors.NewInternalServerError(fmt.Errorf("failed to open input pipe: %w", err))
}
Comment on lines +45 to +76

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== locate exec_support.go and session_exec.go =="
fd -a 'exec_support\.go$|session_exec\.go$|execute\.go$' . | sed 's#^\./##'

echo
echo "== relevant source snippets =="
for f in apps/daemon/pkg/session/exec_support.go apps/daemon/pkg/session/session_exec.go apps/daemon/pkg/session/execute.go; do
  if [ -f "$f" ]; then
    echo "--- $f ($(wc -l < "$f") lines) ---"
    if [ "$f" = "apps/daemon/pkg/session/exec_support.go" ]; then
      sed -n '1,120p' "$f"
    elif [ "$f" = "apps/daemon/pkg/session/session_exec.go" ]; then
      sed -n '1,220p' "$f"
    elif [ "$f" = "apps/daemon/pkg/session/execute.go" ]; then
      sed -n '1,220p' "$f"
    fi
  fi
done

echo
echo "== search for WriteInput/CloseInput usage and wrapper symbols =="
rg -n "WriteInput|CloseInput|cmdWrapperFormat|Start\(|pump|Execute\\(" apps/daemon/pkg -S

Repository: arrrrny/daytona

Length of output: 19389


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== cmdWrapperFormat tail =="
sed -n '200,244p' apps/daemon/pkg/session/execute.go

echo
echo "== toolbox exec controller/session startup and pump/readiness =="
sed -n '1,220p' apps/daemon/pkg/toolbox/process/exec/controller.go
echo "--- session_exec.go ---"
sed -n '1,175p' apps/daemon/pkg/toolbox/process/exec/session_exec.go

echo
echo "== websocket/transport write path around Start/stdin frames =="
rg -n "start|StartFrame|stdin|WriteInput\\(|WriteStdin|Start\\(" apps/daemon/pkg/toolbox/process/exec apps/daemon/pkg/toolbox -S

echo
echo "== FIFO ENXIO behavior probe =="
python3 - <<'PY'
import os, subprocess, time, fcntl, select, stat

pipe = '/tmp/coderabbit_fifo_probe_{}'.format(os.getpid())
try:
    os.mkfifo(pipe)
    # Spawn /bin/cat /dev/null > "$pipe" like the wrapper; wait briefly.
    p = subprocess.Popen(['/bin/bash', '-c', 'cat /dev/null > "$pipe"'], env={'pipe': pipe})
    time.sleep(0.02)
    try:
        fd = os.open(pipe, os.O_WRONLY|os.O_NONBLOCK)
        os.write(fd, b'hello')
        print('open: SUCCESS-write')
        os.close(fd)
    except FileExistsError as e:
        r, _, _ = select.select([p.stdout.fileno()], [], [], 0.5)
        s = p.stdout.readline() or p.stderr.readline() or b''
        print('open: ENXIO?', 'ENXIO' in str(e))
        print('open: errno:', e.errno if hasattr(e, 'errno') else None)
        time.sleep(0.05)
        # Open after reader has attached.
        fd = os.open(pipe, os.O_WRONLY)
        os.write(fd, b'hello')
        os.close(fd)
        print('late_open: SUCCESS-write')
    finally:
        p.terminate()
        p.wait(timeout=2)
finally:
    os.unlink(pipe)
PY

Repository: arrrrny/daytona

Length of output: 49490


Avoid treating an unready FIFO as closed stdin.

WriteInput opens the command’s stdin FIFO with O_WRONLY|O_NONBLOCK and returns "command stdin is closed" on ENXIO. In commandSession.Start, the wrapper pipes stdin through {. cmdfile; } < "$ip" asynchronously; a client that sends stdin immediately after Start() returns can see a spurious Gone error before the command has opened its stdin. Keep the FIFO open from Execute(), or synchronize WriteStdin/WriteInput to the wrapper after stdin has been attached instead of treating ENXIO as an unrecoverable gone state.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/daemon/pkg/session/exec_support.go` around lines 45 - 76, Update
WriteInput and the commandSession.Start/Execute stdin setup so clients cannot
receive a Gone error while the wrapper is still asynchronously attaching the
FIFO; keep the FIFO open from Execute or synchronize WriteInput with stdin
attachment. Preserve the closed-stdin Gone response only after the command has
definitively closed or exited, rather than treating transient ENXIO from
syscall.Open as closure.

defer func() { _ = syscall.Close(fd) }()

// Restore blocking semantics for the write itself so large frames don't
// fail with EAGAIN on a full pipe buffer.
if err := syscall.SetNonblock(fd, false); err != nil {
return common_errors.NewInternalServerError(fmt.Errorf("failed to configure input pipe: %w", err))
}

// write(2) may return fewer bytes than requested (partial write), so loop
// until the whole frame has been delivered.
for remaining := data; len(remaining) > 0; {
n, err := syscall.Write(fd, remaining)
if err != nil {
return common_errors.NewInternalServerError(fmt.Errorf("failed to write to input pipe: %w", err))
}
remaining = remaining[n:]
}

return nil
}

// CloseInput delivers stdin EOF to a running command by tearing down the
// input-holder process that cmdWrapperFormat keeps alive for async commands.
// The holder is a single process (see execute.go), so killing it drops the
// FIFO's last writer and the command's stdin sees EOF — SSH channel EOF
// semantics. Best effort: if the holder is not up yet or already gone, the
// command's stdin stays as-is and nil is returned.
func (s *SessionService) CloseInput(sessionId, commandId string) error {
session, ok := s.sessions.Get(sessionId)
if !ok {
return common_errors.NewNotFoundError(errors.New("session not found"))
}

command, ok := session.commands.Get(commandId)
if !ok {
return common_errors.NewNotFoundError(errors.New("command not found"))
}

if command.ExitCode != nil {
return common_errors.NewGoneError(fmt.Errorf("command has already completed with exit code %d", *command.ExitCode))
}

pidFilePath := filepath.Join(session.Dir(s.configDir), commandId, inputHolderPidFileName)
pidBytes, err := os.ReadFile(pidFilePath)
if err != nil {
return nil
}

pid, err := strconv.Atoi(strings.TrimSpace(string(pidBytes)))
if err != nil || pid <= 0 {
return nil
}

// Kill the holder (and any descendants, defensively) so no process keeps
// the FIFO's write end open.
_ = s.signalProcessTree(pid, syscall.SIGKILL)
if holder, err := os.FindProcess(pid); err == nil {
_ = holder.Signal(syscall.SIGKILL)
}

// The pid file is single-use: remove it so a later CloseInput doesn't
// signal a recycled PID.
_ = os.Remove(pidFilePath)

return nil
}

// SignalDescendants delivers sig to every descendant of the session's shell
// process — i.e. the currently running command pipeline (command subshell,
// labelers, stdin holder) — without touching the shell itself, so the wrapper
// survives to record the command's exit code (e.g. 130 for SIGINT).
func (s *SessionService) SignalDescendants(sessionId string, sig syscall.Signal) error {
session, ok := s.sessions.Get(sessionId)
if !ok {
return common_errors.NewNotFoundError(errors.New("session not found"))
}

if session.cmd == nil || session.cmd.Process == nil {
return common_errors.NewGoneError(errors.New("session process is not running"))
}

if session.cmd.ProcessState != nil && session.cmd.ProcessState.Exited() {
return common_errors.NewGoneError(errors.New("session process has exited"))
}

return s.signalProcessTree(session.cmd.Process.Pid, sig)
}
9 changes: 8 additions & 1 deletion apps/daemon/pkg/session/execute.go
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,11 @@ func (s *SessionService) Execute(sessionId, cmdId, cmd string, async, isCombined

inputPipeCommand := `cat /dev/null > "$ip" &`
if async {
inputPipeCommand = `while :; do sleep 3600; done > "$ip" &`
// The holder must be a single process: killing the recorded PID alone
// must drop the FIFO's last writer so CloseInput delivers EOF
// immediately. A `while :; do sleep; done` loop leaves its sleep child
// holding the FIFO open after the loop shell is killed.
inputPipeCommand = `exec tail -f /dev/null > "$ip" &`
}

cmdToExec := fmt.Sprintf(cmdWrapperFormat+"\n",
Expand Down Expand Up @@ -216,6 +220,9 @@ var cmdWrapperFormat string = `
%s
ip_pid=$!

# Record the input-holder PID so CloseInput can deliver stdin EOF later.
echo "$ip_pid" > "$dir/input_holder.pid" 2>/dev/null || true

# Run your command from file (avoids heredoc parsing issues with pipe-fed shells)
{ . %q; } < "$ip" > "$sp" 2> "$ep"
_ec=$?
Expand Down
Loading