Shell command logging and streaming

[adamhjk]

When running shell commands from withing a models methods, it should log what it is doing and optionally stream stdout/stderr in addition to any other data storage requirements.

[adamhjk]

Design: executeProcess Infrastructure Utility

Problem

Shell command execution is duplicated across 5+ files in the codebase. Each reimplements: line-by-line stream parsing, TextDecoder buffer management, timeout handling, and error capture. None integrate with LogTape automatically.

Solution

A shared infrastructure utility at src/infrastructure/process/process_executor.ts that wraps Deno.Command with:

1. Optional line-by-line streaming of stdout/stderr through callbacks
2. Optional LogTape logger integration (stdout → info, stderr → warn)
3. Uniform timeout handling
4. Captured stdout, stderr, exit code, and duration

API

export type LineCallback = (line: string) => void;

export interface ProcessExecutorOptions {
  command: string;
  args?: string[];
  cwd?: string;
  env?: Record<string, string>;
  timeoutMs?: number;
  streaming?: {
    onStdout?: LineCallback;
    onStderr?: LineCallback;
    logger?: Logger;  // stdout → info, stderr → warn
  };
}

export interface ProcessResult {
  exitCode: number;
  success: boolean;
  stdout: string;
  stderr: string;
  durationMs: number;
}

export async function executeProcess(
  options: ProcessExecutorOptions,
): Promise<ProcessResult>;

Behavior

• Streaming (when streaming is provided): spawns process, streams stdout/stderr line-by-line via streamLines helper, calling callbacks + logger per line
• Buffered (when streaming is omitted): uses command.output()
• Timeout: buffered mode uses AbortSignal.timeout(), streaming mode races process.status against a timeout that kills with SIGTERM
• Error handling: process startup failure → exitCode = -1

DDD Placement

Infrastructure concern at src/infrastructure/process/process_executor.ts, following the pattern of stdin_reader.ts and editor_service.ts. Domain models consume it without knowing about LogTape internals. The existing MethodStreamingCallbacks interface is structurally compatible with no changes needed.

Usage: Model Method

const result = await executeProcess({
  command: "sh",
  args: ["-c", attrs.run],
  cwd: attrs.workingDir,
  env: attrs.env,
  timeoutMs: attrs.timeout,
  streaming: (context.streaming?.onStdout || context.streaming?.onStderr)
    ? {
        onStdout: context.streaming?.onStdout,
        onStderr: context.streaming?.onStderr,
      }
    : undefined,
});

Usage: Workflow Step with LogTape

const result = await executeProcess({
  command: task.command,
  args: task.args,
  cwd: task.workingDir ?? ctx.repoDir,
  env: task.env,
  streaming: shouldStream
    ? {
        onStdout: (line) => {
          if (ctx.progress?.onStepStdout && ctx.workflowRun) {
            ctx.progress.onStepStdout(ctx.workflowRun, ctx.jobName, ctx.stepName, line);
          }
        },
        onStderr: (line) => {
          if (ctx.progress?.onStepStderr && ctx.workflowRun) {
            ctx.progress.onStepStderr(ctx.workflowRun, ctx.jobName, ctx.stepName, line);
          }
        },
        logger: stepLogger,
      }
    : undefined,
});

Files to Refactor

File	Change
keeb/shell/shell_model.ts	Remove streamOutput(), replace 3-branch if/else with single executeProcess() call
workflows/execution_service.ts	Remove streamOutput(), collapse executeShell + executeShellStreaming
aws/cli/aws_cli_model.ts	Replace manual AbortController/setTimeout (~40 lines → ~5)
systemd/journalctl/journalctl_model.ts	Replace manual reader loop (~60 lines → ~10)
Extension model skill docs	Update shell command example to use executeProcess

[adamhjk]

Updated Design: Shell Command Execution with LogTape Streaming

This supersedes the previous comment. The design has evolved to use LogTape file sinks for persistence instead of callbacks.

Problem

Shell command execution is duplicated across 5+ files. Each reimplements streaming, buffering, timeout handling, and error capture. Stdout/stderr persistence uses a fragile callback chain: streaming callbacks → onStepStdout/onStepStderr progress callbacks → StepRun.appendStdout() → throttled YAML save. This embeds stdout/stderr in the workflow run YAML, which is read back by the LogStreamService for the TUI.

PR #229 added logger: Logger to MethodContext. Every model method now has a pre-configured LogTape logger. We should use this as the single interface for all output.

Goal: Replace the callback chain with LogTape file sinks. All process output flows through the logger. File sinks persist logs to disk. No callbacks.

New Component: RunFileSink

A custom LogTape sink that dynamically routes log records to per-run log files. Since LogTape can only be configure()d once, we register a single sink at init time and dynamically add/remove file targets at runtime.

src/infrastructure/logging/run_file_sink.ts

How it works:

• At init, one RunFileSink instance is created and its sink registered with LogTape for categories ["model", "method", "run"] and ["workflow", "run"]
• When a model method run starts, caller does runFileSink.register(["model", "method", "run", name, method], logFilePath)
• Every log record whose category starts with that prefix gets written to the file (formatted as text)
• When the run completes, runFileSink.unregister(...) flushes and closes
• The sink matches the longest registered prefix for each record

Log File Locations

.swamp/outputs/{type}/{method}/{def-id}-{timestamp}.log    # model method runs
.swamp/workflow-runs/{workflowId}/workflow-run-{runId}.log  # workflow runs

executeProcess API

export interface ProcessExecutorOptions {
  command: string;
  args?: string[];
  cwd?: string;
  env?: Record<string, string>;
  timeoutMs?: number;
  /** Logger for streaming stdout (info) and stderr (warn). */
  logger?: Logger;
}

export interface ProcessResult {
  exitCode: number;
  success: boolean;
  stdout: string;
  stderr: string;
  durationMs: number;
}

export async function executeProcess(
  options: ProcessExecutorOptions,
): Promise<ProcessResult>;

No onStdout/onStderr callbacks. No streaming option. Just pass a logger. When logger is provided, output is streamed line-by-line through it and still captured in the returned ProcessResult for further processing. When omitted, buffered execution.

Usage

// Model method (the common case)
const result = await executeProcess({
  command: "sh",
  args: ["-c", attrs.run],
  cwd: attrs.workingDir,
  env: attrs.env,
  timeoutMs: attrs.timeout,
  logger: context.logger,
});
// Console output + file persistence handled by LogTape.
// result.stdout/stderr still available for parsing.

// Workflow step execution
const result = await executeProcess({
  command: task.command,
  args: task.args,
  cwd: task.workingDir ?? ctx.repoDir,
  logger: stepLogger,
});

// Buffered (no streaming)
const result = await executeProcess({
  command: "tar",
  args: ["-xzf", tarballPath, "-C", tempDir],
});

What Gets Removed

• onStepStdout/onStepStderr from ExecutionProgressCallback
• appendStdout()/appendStderr()/_stdoutLines/_stderrLines/_updateOutput() from StepRun
• createThrottledSave() mechanism
• streamOutput() duplicated in shell_model.ts and execution_service.ts
• Callback wrapping logic in executeJob()

Files to Create

File	Purpose
src/infrastructure/process/process_executor.ts	executeProcess + streamLines
src/infrastructure/process/process_executor_test.ts	Unit tests
src/infrastructure/logging/run_file_sink.ts	RunFileSink
src/infrastructure/logging/run_file_sink_test.ts	Unit tests

Files to Modify

File	Change
src/infrastructure/logging/logger.ts	Register global RunFileSink instance
src/domain/models/keeb/shell/shell_model.ts	Replace streamOutput() with executeProcess()
src/domain/workflows/execution_service.ts	Remove callbacks/throttled save, use executeProcess + logger
src/domain/workflows/workflow_run.ts	Remove stdout/stderr accumulation from StepRun
src/presentation/output/components/workflow_execution/LogStreamService.ts	Read .log files instead of embedded YAML
src/cli/commands/model_method_run.ts	Register/unregister file sink targets
src/domain/models/aws/cli/aws_cli_model.ts	Replace manual timeout with executeProcess()
src/domain/models/systemd/journalctl/journalctl_model.ts	Replace manual reader loop with executeProcess()
.claude/skills/swamp-extension-model/SKILL.md	Update shell command example
.claude/skills/swamp-extension-model/references/examples.md	Add executeProcess example
Tests	Update to remove callback assertions

[adamhjk]

just merged the fix!