Skip to main content

JobManager

Struct JobManager 

Source
pub struct JobManager { /* private fields */ }
Expand description

Manager for background jobs.

Implementations§

Source§

impl JobManager

Source

pub fn new() -> Self

Create a new job manager.

On construction, best-effort prunes stale job output files left by previously crashed kaish processes. All errors are intentionally ignored — startup cleanup is opportunistic and must never prevent the manager from being created (silent-fallback rule: the only case where silent is correct is read-only / cleanup-only paths with no data loss risk).

§Scoping decision

All sessions share a single /tmp/kaish/jobs/ directory. Filenames embed the OS PID that wrote them (session_S_job_J.PID.txt). Files from the current process are never touched here — only files whose embedded PID refers to a dead process are removed. On Linux we check /proc/{pid} for existence; on other platforms we skip the prune rather than guess.

Source

pub fn set_kill_grace(&self, grace: Duration)

Mirror KernelConfig::kill_grace onto the manager (see the field doc).

Source

pub fn kill_grace(&self) -> Duration

The cancellation cascade’s SIGTERM→SIGKILL grace (see Self::set_kill_grace).

Source

pub fn set_finished_retention(&self, keep: u64)

Set how many finished jobs stay tracked (default 100, DEFAULT_FINISHED_RETENTION). 0 keeps no finished jobs beyond the gate-safety rule — gated and stopped jobs are never evicted regardless.

Source

pub fn set_persist_output_files(&self, on: bool)

Toggle whether completed jobs persist their output to a host temp file.

Disable this for a hermetic / read-only kernel: the host write in Job::write_output_file uses std::fs directly and so bypasses the VFS (and any read-only mount). Turning it off costs a hermetic kernel nothing it cannot get elsewhere — the job’s output is in its live streams (/v/jobs/{id}/stdout, or JobManager::read_stdout), bounded by a 10 MB ring. Redirect to a VFS path (cmd > /tmp/out &) when a job outruns that ring.

Must be set before jobs are spawned — the flag is stamped onto each job at registration time, not consulted at completion.

Source

pub fn persist_output_files(&self) -> bool

Whether completed jobs persist their output to a host temp file.

Source

pub async fn spawn<F>(&self, command: String, future: F) -> JobId
where F: Future<Output = ExecResult> + Send + 'static,

Spawn a new background job from a future.

The job is inserted into the map synchronously before returning, guaranteeing it’s immediately queryable via exists() or get().

Source

pub async fn register(&self, command: String, rx: Receiver<ExecResult>) -> JobId

Spawn a job that’s already running and communicate via channel.

Source

pub async fn streams(&self, id: JobId) -> Option<JobStreams>

A job’s live output streams, or None if there is no such job.

The producer side: Kernel::try_execute_external takes these for the job it is running under and tees the child’s pipes into them as the bytes arrive.

Source

pub async fn read_stdout(&self, id: JobId) -> Option<Vec<u8>>

Snapshot a job’s stdout so far, or None if there is no such job.

Readable while the job runs — that is the point. None and Some(vec![]) are different answers: no such job, versus a job that has not written anything yet.

Source

pub async fn read_stderr(&self, id: JobId) -> Option<Vec<u8>>

Snapshot a job’s stderr so far, or None if there is no such job. See Self::read_stdout.

Source

pub async fn finalize_streams(&self, id: JobId, result: &ExecResult)

Close out a finished job’s streams: write the captured result into a stream that received nothing live, then close both.

The conditional is the no-double-write rule. A stream with live bytes in it already holds exactly what the child emitted; writing result.text_out() on top would repeat all of it. A stream with no live bytes belongs to a job with nothing to tee — a builtin returns its output as a value, not as a pipe — and would otherwise read empty forever.

Called by the background task that owns the job, before it hands the result over, so a reader that sees a terminal status also sees a closed, complete stream.

Source

pub async fn wait(&self, id: JobId) -> Option<ExecResult>

Wait for a specific job to complete.

Returns None when the job does not exist or is stopped — a stopped job can never finish, so waiting on one would hang forever. Callers that need to tell the two apart check JobManager::get after a None.

The job’s pending awaitable (its task handle or result channel) is taken out of the map under the lock, then the lock is released before awaiting completion. Holding the jobs mutex across the await would block every other job operation (spawn/register/list/status/kill) for the whole duration of the job — so a nested & started under a parked wait %N would deadlock. The lock is re-acquired only to finalize (persist output, cache the result).

Source

pub async fn wait_all(&self) -> Vec<(JobId, ExecResult)>

Wait for all jobs that can still finish, returning results in completion order.

Stopped jobs are skipped, and that is load-bearing. A Ctrl-Z’d job is registered by JobManager::register_stopped with no JoinHandle and no result channel, and Job::is_done returns false for as long as it is stopped — so nothing can ever make it done. Waiting on one here spun the 10ms poll loop forever, and since crate::Kernel::shutdown calls this, a single Ctrl-Z hung shutdown with no timeout and no escape. Skip them: wait_all means “wait for everything that will finish”, not “wait for everything”.

The filter below is a snapshot; a job that stops after it (the bg reaper observing a SIGSTOP) is caught by JobManager::wait’s own stopped guard, which returns None instead of re-creating the hang.

A caller that wants a stopped job to finish must resume it first (bg/fg).

Source

pub async fn list(&self) -> Vec<JobInfo>

List all jobs with their status.

Listing polls every job, so this is also a completion-observation point: retention is enforced here (after the snapshot is taken — the returned list is complete even for entries evicted by it).

Sorted by JobId (GH #247) — the backing map is a HashMap, whose iteration order is arbitrary and was leaking straight through to jobs, /v/jobs, and --json: two jobs could list as [2, 1]. An MCP caller handed that order, or a snapshot test pinned against it, saw a flake with no code change — sorting makes the order a stated contract instead of whatever the hasher happened to do.

Source

pub async fn running_count(&self) -> usize

Get the number of running jobs.

Source

pub async fn reap_finished(&self) -> Vec<JobInfo>

Remove completed jobs from tracking and clean up their temp files, returning info for each job removed.

Shared by jobs --cleanup (which only needs a count) and the REPL’s pre-prompt notification (GH #131, which needs the id/command/status of each job so it can print [N]+ Done ... before reaping it) — one rule for “is this job safe to reap”, not two copies that could drift.

Source

pub async fn cleanup(&self)

Remove completed jobs from tracking and clean up their temp files.

The count-only form of reap_finished that jobs --cleanup reports.

Source

pub async fn exists(&self, id: JobId) -> bool

Check if a specific job exists.

Source

pub async fn get(&self, id: JobId) -> Option<JobInfo>

Get info for a specific job.

Source

pub async fn get_command(&self, id: JobId) -> Option<String>

Get the command string for a job.

Source

pub async fn get_status_string(&self, id: JobId) -> Option<String>

Get the status string for a job (for /v/jobs/{id}/status).

Source

pub async fn list_ids(&self) -> Vec<JobId>

List all job IDs, sorted (GH #247 — see Self::list’s doc for why the backing HashMap’s iteration order is not good enough here: this backs the /v/jobs directory listing via crate::vfs::JobFs).

Source

pub async fn register_stopped( &self, command: String, pid: u32, pgid: u32, ) -> JobId

Register a stopped job (from Ctrl-Z on a foreground process).

Source

pub async fn stop_job(&self, id: JobId, pid: u32, pgid: u32)

Mark a job as stopped with its process info.

Source

pub async fn resume_job(&self, id: JobId)

Mark a stopped job as resumed.

Source

pub async fn last_stopped(&self) -> Option<JobId>

Get the most recently stopped job.

Source

pub async fn get_process_info(&self, id: JobId) -> Option<(u32, u32)>

Get process info (pid, pgid) for a job.

Source

pub async fn set_cancel_token(&self, id: JobId, token: CancellationToken)

Record the cancellation token of the fork running a background job, so kill %N can stop the job even when it has no OS process group of its own (e.g. a pure builtin like sleep &).

Source

pub async fn mark_killed_and_cancel(&self, id: JobId, delivered: bool) -> bool

Flag a terminating kill and trip the job’s cancellation token, as one operation under the jobs lock. Returns false — and leaves the job unflagged — when there is no lever to kill with: no cancellation token recorded and no OS signal already delivered (delivered). The flag turns the job’s terminal status into Killed, so setting it without a working delivery would misclassify a later organic failure as a kill (found in review: JobManager::spawn/register jobs have no token unless the kernel records one).

The flag is set before the token trips (the job can unwind the instant it does; a flag set after races the status read), and the token is cancelled after the lock drops — CancellationToken::cancel is synchronous, but waking waiters under the jobs lock buys nothing.

Source

pub async fn cancel(&self, id: JobId) -> bool

Cancel a job by its token. Returns true if a token was recorded and cancelled. The cancellation cascade stops in-process builtin futures and SIGTERM→SIGKILLs any external children’s process groups.

Source

pub async fn add_pgid(&self, id: JobId, pgid: u32)

Record a process group spawned while running a background job. Lets kill -<sig> %N deliver an arbitrary signal directly to the real processes. Deduplicated (a job may spawn several externals).

Source

pub async fn job_pgids(&self, id: JobId) -> Vec<u32>

The process groups recorded for a job (empty for a pure-builtin job). Includes the legacy single pgid recorded for stopped jobs (Ctrl-Z), so kill %N signals a stopped foreground job’s group too.

Source

pub async fn try_result(&self, id: JobId) -> Option<ExecResult>

Non-blocking accessor for a finished job’s result — None while the job is still Running/Stopped, or if id doesn’t exist. Unlike Self::wait, this never parks: it polls once and returns whatever is (or isn’t) already available. GH #243: previously the only ways to read a job’s ExecResult were wait (blocks until done) or string-parsing failed:{code} off /v/jobs/N/status.

Source

pub async fn remove(&self, id: JobId)

Remove a job from tracking.

Trait Implementations§

Source§

impl Default for JobManager

Source§

fn default() -> Self

Returns the “default value” for a type. Read more

Auto Trait Implementations§

Blanket Implementations§

Source§

impl<T> Any for T
where T: 'static + ?Sized,

Source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
Source§

impl<T> Borrow<T> for T
where T: ?Sized,

Source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
Source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
Source§

impl<T> From<T> for T

Source§

fn from(t: T) -> T

Returns the argument unchanged.

Source§

impl<T> FutureExt for T

Source§

fn with_context(self, otel_cx: Context) -> WithContext<Self>

Attaches the provided Context to this type, returning a WithContext wrapper. Read more
Source§

fn with_current_context(self) -> WithContext<Self>

Attaches the current Context to this type, returning a WithContext wrapper. Read more
Source§

impl<T> Instrument for T

Source§

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
Source§

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
Source§

impl<T, U> Into<U> for T
where U: From<T>,

Source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Source§

impl<'src, T> IntoMaybe<'src, T> for T
where T: 'src,

Source§

type Proj<U: 'src> = U

Source§

fn map_maybe<R>( self, _f: impl FnOnce(&'src T) -> &'src R, g: impl FnOnce(T) -> R, ) -> <T as IntoMaybe<'src, T>>::Proj<R>
where R: 'src,

Source§

impl<T> Pointable for T

Source§

const ALIGN: usize

The alignment of pointer.
Source§

type Init = T

The type for initializers.
Source§

unsafe fn init(init: <T as Pointable>::Init) -> usize

Initializes a with the given initializer. Read more
Source§

unsafe fn deref<'a>(ptr: usize) -> &'a T

Dereferences the given pointer. Read more
Source§

unsafe fn deref_mut<'a>(ptr: usize) -> &'a mut T

Mutably dereferences the given pointer. Read more
Source§

unsafe fn drop(ptr: usize)

Drops the object pointed to by the given pointer. Read more
Source§

impl<T> Same for T

Source§

type Output = T

Should always be Self
Source§

impl<T, S> SpanWrap<S> for T
where S: WrappingSpan<T>,

Source§

fn with_span(self, span: S) -> <S as WrappingSpan<Self>>::Spanned

Invokes WrappingSpan::make_wrapped to wrap an AST node in a span.
Source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

Source§

type Error = Infallible

The type returned in the event of a conversion error.
Source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
Source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

Source§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
Source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
Source§

impl<V, T> VZip<V> for T
where V: MultiLane<T>,

Source§

fn vzip(self) -> V

Source§

impl<T> WithSubscriber for T

Source§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
Source§

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more