feat(core): tasks on the machine tree, and tty7 task

A workspace now keeps a board: `Workspace::tasks`, each a title, a prompt,
where to run it, which agent, and the runs started for it. Tasks ride the
machine tree like the sidebar groups do — persisted with it, moved as
`TaskPut` / `TaskRemoved` layout deltas, gone with the workspace.

A task stores no column. `task::column` reads it off the agents in its runs'
tabs (Queued, Running, Needs input, Review, Done); the only verdict kept is
the user's "done", and it yields when an agent takes another turn.

Compatibility: a client now names what it can decode in
`ControlHello::features`, and a server holds the board's deltas back from a
client that did not say `tasks` — an event an older GUI cannot decode would
end its link. The server advertises `tasks` so a newer client never sends
`TaskPut` to an older server.

`tty7 task ls|add|done|reopen|rm` works the board from the CLI; the bundled
skill documents it.
This commit is contained in:
l0ng-ai
2026-09-28 23:07:41 +08:00
parent 4f8237f361
commit e5907fdec1
10 changed files with 1128 additions and 5 deletions
+5 -1
View File
@@ -189,7 +189,11 @@ pub mod mock {
build: "mock".into(),
separator: '\\',
home: "C:\\Users\\mock".into(),
features: vec![feature::CONTROL.into(), feature::MACHINE_TREE.into()],
features: vec![
feature::CONTROL.into(),
feature::MACHINE_TREE.into(),
feature::TASKS.into(),
],
instance: "mock-instance".into(),
})
}
+68
View File
@@ -147,6 +147,12 @@ pub enum Command {
#[command(subcommand, about = "Panes: the terminals themselves (%N addresses)")]
Pane(PaneCmd),
#[command(
subcommand,
about = "The board: tasks for agents, and where each one stands"
)]
Task(TaskCmd),
#[command(
subcommand,
about = "Machines: this one plus every remote the server has a link to"
@@ -575,6 +581,68 @@ pub enum TabCmd {
},
}
#[derive(Debug, Subcommand)]
pub enum TaskCmd {
#[command(about = "Tasks on a workspace's board, with the column each one is in")]
Ls {
#[arg(value_name = "WORKSPACE")]
ws: Option<String>,
},
#[command(
about = "Put a task on the board (queued; start it from the GUI's board)",
long_about = "Put a task on a workspace's board, queued.\n\n\
The GUI's board starts it: a new tab in --cwd, running --agent, \
told the prompt (or the title, when there is no prompt). \
Prints the task's id."
)]
Add {
#[arg(value_name = "TITLE")]
title: String,
#[arg(
long,
value_name = "TEXT",
help = "What the agent is told; defaults to the title"
)]
prompt: Option<String>,
#[arg(
long,
value_name = "FILE",
conflicts_with = "prompt",
help = "Read the prompt from a file ('-' for stdin)"
)]
prompt_file: Option<String>,
#[arg(long, value_name = "DIR", help = "Where the agent runs")]
cwd: Option<String>,
#[arg(
long,
value_name = "AGENT",
help = "Which agent runs it, e.g. claude, codex"
)]
agent: Option<String>,
#[arg(long, value_name = "WORKSPACE")]
ws: Option<String>,
},
#[command(about = "Mark a task finished")]
Done {
#[arg(value_name = "TASK")]
task: String,
},
#[command(about = "Take a task out of Done")]
Reopen {
#[arg(value_name = "TASK")]
task: String,
},
#[command(about = "Delete a task from the board")]
Rm {
#[arg(value_name = "TASK")]
task: String,
},
}
#[derive(Debug, Subcommand)]
pub enum PaneCmd {
#[command(about = "Panes of a workspace, or of the whole machine")]
+376 -1
View File
@@ -12,7 +12,7 @@ use crate::address::{self, Context, WorkspaceAddress};
use crate::backend::{Backend, RunSpec};
use crate::cli::{
CaptureArgs, Cli, Command, ExecArgs, MachineCmd, PaneCmd, RunArgs, SendArgs, ServerCmd,
SplitArgs, TabCmd, WaitArgs, WaitState, WsCmd,
SplitArgs, TabCmd, TaskCmd, WaitArgs, WaitState, WsCmd,
};
use crate::exec::ExecEnd;
use crate::output;
@@ -89,6 +89,32 @@ pub fn execute(cli: Cli, ctx: &Context, backend: &mut dyn Backend) -> Result<Out
Some(Command::Tab(TabCmd::Close { tab })) => tab_close(&tab, backend),
Some(Command::Tab(TabCmd::Rename { tab, name })) => tab_rename(&tab, name, backend),
Some(Command::Tab(TabCmd::Move { tab, index })) => tab_move(&tab, index, backend),
Some(Command::Task(TaskCmd::Ls { ws })) => task_ls(ws.as_deref(), ctx, backend),
Some(Command::Task(TaskCmd::Add {
title,
prompt,
prompt_file,
cwd,
agent,
ws,
})) => {
let prompt = match prompt_file {
Some(path) => Some(read_prompt_file(&path)?),
None => prompt,
};
task_add(
title,
prompt,
cwd,
agent.as_deref(),
ws.as_deref(),
ctx,
backend,
)
}
Some(Command::Task(TaskCmd::Done { task })) => task_mark(&task, true, backend),
Some(Command::Task(TaskCmd::Reopen { task })) => task_mark(&task, false, backend),
Some(Command::Task(TaskCmd::Rm { task })) => task_rm(&task, backend),
Some(Command::Pane(PaneCmd::Ls { ws, all })) => pane_ls(ws.as_deref(), all, backend),
Some(Command::Pane(PaneCmd::Close { targets, orphans })) => {
pane_close(&targets, orphans, ctx, backend)
@@ -1138,6 +1164,227 @@ fn tab_move(tab: &str, index: u64, backend: &mut dyn Backend) -> Result<Outcome>
report("", json!({ "tab": tab.to_string(), "to": index }))
}
/// Refuses up front on a server that keeps no board. It would refuse the
/// request anyway, but as an unknown variant — which says nothing about what
/// to do.
fn require_board(backend: &mut dyn Backend) -> Result<()> {
if !backend
.hello()?
.has_feature(tty7_core::daemon::control::feature::TASKS)
{
bail!("this server keeps no board — update it (`tty7 server restart`) to use tasks");
}
Ok(())
}
/// What each pane's agent last reported, by pane id. A server that cannot say
/// leaves every run looking finished, which is the safe misreading: a task
/// shows in Review rather than claiming to be busy.
fn agent_states_by_pane(
backend: &mut dyn Backend,
) -> std::collections::HashMap<u64, tty7_core::core::cli_agent::AgentSessionState> {
match backend.control(ControlRequest::AgentStates) {
Ok(ReplyOk::AgentStates(states)) => {
states.into_iter().map(|s| (s.pane_id, s.state)).collect()
}
_ => Default::default(),
}
}
/// The live agent a run's tab is running, read the way the GUI's board reads
/// it: the tab's most urgent agent pane speaks for it.
fn run_live(
ws: &Workspace,
run: &tty7_core::core::task::Run,
states: &std::collections::HashMap<u64, tty7_core::core::cli_agent::AgentSessionState>,
) -> Option<tty7_core::core::task::Live> {
use tty7_core::core::cli_agent::AgentStatus;
let tab = ws.tabs.iter().find(|t| Some(t.id) == run.tab)?;
let urgency = |s: AgentStatus| match s {
AgentStatus::Waiting => 3,
AgentStatus::Working => 2,
AgentStatus::Done => 1,
AgentStatus::Idle => 0,
};
tab.root
.pane_ids()
.iter()
.filter_map(|p| states.get(p))
.max_by_key(|s| urgency(s.status))
.map(tty7_core::core::task::Live::from)
}
fn column_name(column: tty7_core::core::task::Column) -> &'static str {
use tty7_core::core::task::Column;
match column {
Column::Queued => "queued",
Column::Running => "running",
Column::NeedsInput => "needs-input",
Column::Review => "review",
Column::Done => "done",
}
}
fn task_ls(explicit: Option<&str>, ctx: &Context, backend: &mut dyn Backend) -> Result<Outcome> {
let machine = fetch_machine(backend)?;
let id = resolve_ws(explicit, ctx, &machine)?;
let ws = machine
.workspaces
.iter()
.find(|ws| ws.id == id)
.expect("resolve_ws returned an id straight out of this machine");
let states = agent_states_by_pane(backend);
let columns: Vec<_> = ws
.tasks
.iter()
.map(|task| tty7_core::core::task::column(task, |run| run_live(ws, run, &states)))
.collect();
let rows: Vec<Vec<String>> = ws
.tasks
.iter()
.zip(&columns)
.map(|(task, column)| {
vec![
short_task_id(task.id),
column_name(*column).to_string(),
task.runs
.last()
.map(|r| r.agent)
.or(task.agent)
.map_or("-".to_string(), |a| a.slug().to_string()),
task.title.clone(),
]
})
.collect();
let tasks: Vec<Value> = ws
.tasks
.iter()
.zip(&columns)
.map(|(task, column)| {
json!({
"id": task.id.to_string(),
"title": task.title,
"prompt": task.prompt,
"column": column_name(*column),
"cwd": task.cwd,
"agent": task.agent.map(|a| a.slug()),
"runs": task.runs.iter().map(|r| json!({
"agent": r.agent.slug(),
"tab": r.tab.map(|t| t.to_string()),
"session_id": r.session_id,
"started": r.started,
})).collect::<Vec<_>>(),
"created": task.created,
"done": task.done.map(|d| d.at),
})
})
.collect();
report(
output::table(&["TASK", "COLUMN", "AGENT", "TITLE"], &rows),
json!({ "workspace": id.to_string(), "tasks": tasks }),
)
}
/// The first eight hex digits — as much of an id as a person types, and
/// what [`find_task`] takes back.
fn short_task_id(id: tty7_core::core::task::TaskId) -> String {
id.to_string().chars().take(8).collect()
}
fn read_prompt_file(path: &str) -> Result<String> {
if path == "-" {
let mut text = String::new();
std::io::Read::read_to_string(&mut std::io::stdin(), &mut text)
.context("reading the prompt from stdin")?;
return Ok(text);
}
std::fs::read_to_string(path).with_context(|| format!("reading the prompt from {path}"))
}
fn task_add(
title: String,
prompt: Option<String>,
cwd: Option<String>,
agent: Option<&str>,
explicit: Option<&str>,
ctx: &Context,
backend: &mut dyn Backend,
) -> Result<Outcome> {
require_board(backend)?;
let machine = fetch_machine(backend)?;
let workspace = resolve_ws(explicit, ctx, &machine)?;
let mut task = tty7_core::core::task::Task::new(title.trim());
task.prompt = prompt.unwrap_or_default().trim().to_string();
task.cwd = cwd;
task.agent = match agent {
Some(name) => Some(
tty7_core::core::cli_agent::CLIAgent::from_slug(name)
.with_context(|| format!("no agent is called {name:?}"))?,
),
None => None,
};
if let Err(why) = task.check() {
bail!(why);
}
let id = task.id;
backend.control(ControlRequest::TaskPut { workspace, task })?;
report(
id.to_string(),
json!({ "workspace": workspace.to_string(), "task": id.to_string() }),
)
}
/// The one task on this machine whose id is, or starts with, `text`.
fn find_task<'a>(
machine: &'a Machine,
text: &str,
) -> Result<(&'a Workspace, &'a tty7_core::core::task::Task)> {
let text = text.trim().to_ascii_lowercase();
if text.len() < 4 {
bail!("give at least four characters of the task id");
}
let mut found = machine.workspaces.iter().flat_map(|ws| {
ws.tasks
.iter()
.filter(|t| t.id.to_string().starts_with(&text))
.map(move |t| (ws, t))
});
match (found.next(), found.next()) {
(Some(hit), None) => Ok(hit),
(None, _) => bail!("no task {text} on this machine (tty7 task ls lists them)"),
(Some(_), Some(_)) => bail!("{text} names more than one task; give more of the id"),
}
}
fn task_mark(text: &str, done: bool, backend: &mut dyn Backend) -> Result<Outcome> {
require_board(backend)?;
let machine = fetch_machine(backend)?;
let (ws, task) = find_task(&machine, text)?;
let states = agent_states_by_pane(backend);
let mut task = task.clone();
task.done = done.then(|| tty7_core::core::task::Done {
at: tty7_core::core::machine::unix_now(),
turns: tty7_core::core::task::live_turns(&task, |run| run_live(ws, run, &states)),
});
let id = task.id;
backend.control(ControlRequest::TaskPut {
workspace: ws.id,
task,
})?;
report("", json!({ "task": id.to_string(), "done": done }))
}
fn task_rm(text: &str, backend: &mut dyn Backend) -> Result<Outcome> {
require_board(backend)?;
let machine = fetch_machine(backend)?;
let (ws, task) = find_task(&machine, text)?;
backend.control(ControlRequest::TaskRemove {
workspace: ws.id,
task: task.id,
})?;
report("", json!({ "task": task.id.to_string(), "removed": true }))
}
fn pane_ls(explicit: Option<&str>, all: bool, backend: &mut dyn Backend) -> Result<Outcome> {
if all {
return pane_ls_all(backend);
@@ -2282,6 +2529,134 @@ mod tests {
);
}
#[test]
fn task_add_puts_a_queued_task_on_the_named_workspace() {
let mut backend = mock();
let api = backend.machine.workspaces[0].id;
let out = run_cli(
&[
"tty7",
"task",
"add",
" Fix the flaky test ",
"--prompt",
"it fails 1 in 10",
"--agent",
"codex",
"--cwd",
"/src/api",
"--ws",
"api",
],
&Context::default(),
&mut backend,
);
let Some(ControlRequest::TaskPut { workspace, task }) = backend.control_calls.last() else {
panic!("expected a TaskPut, got {:?}", backend.control_calls);
};
assert_eq!(*workspace, api);
assert_eq!(task.title, "Fix the flaky test");
assert_eq!(task.prompt, "it fails 1 in 10");
assert_eq!(task.agent, Some(CLIAgent::Codex));
assert_eq!(task.cwd.as_deref(), Some("/src/api"));
assert!(task.runs.is_empty() && task.done.is_none());
assert_eq!(
human(out),
task.id.to_string(),
"the id is the printed result"
);
}
#[test]
fn task_add_refuses_an_agent_nobody_has_heard_of() {
let mut backend = mock();
let err = execute(
cli(&["tty7", "task", "add", "x", "--agent", "hal", "--ws", "api"]),
&Context::default(),
&mut backend,
)
.unwrap_err();
assert!(err.to_string().contains("hal"), "{err}");
assert!(
!backend
.control_calls
.iter()
.any(|c| matches!(c, ControlRequest::TaskPut { .. }))
);
}
#[test]
fn task_ls_reads_each_column_off_the_agent_in_the_run_tab() {
use tty7_core::core::cli_agent::{AgentSessionState, AgentStatus};
use tty7_core::core::task::{Run, Task};
let mut backend = mock();
let ws = &mut backend.machine.workspaces[0];
let build_tab = ws.tabs[0].id;
let queued = Task::new("write the docs");
let mut running = Task::new("fix the build");
running.push_run(Run {
agent: CLIAgent::Claude,
tab: Some(build_tab),
session_id: None,
started: 0,
});
ws.tasks = vec![queued, running];
backend.replies.push_back(ReplyOk::AgentStates(vec![
tty7_core::daemon::control::PaneAgentState {
pane_id: 1,
agent: Some(CLIAgent::Claude),
state: AgentSessionState {
status: AgentStatus::Waiting,
..AgentSessionState::default()
},
},
]));
let out = json_of(run_cli(
&["tty7", "task", "ls", "api"],
&Context::default(),
&mut backend,
));
let columns: Vec<&str> = out["tasks"]
.as_array()
.unwrap()
.iter()
.map(|t| t["column"].as_str().unwrap())
.collect();
assert_eq!(columns, vec!["queued", "needs-input"]);
}
#[test]
fn task_done_finds_the_task_by_the_start_of_its_id() {
use tty7_core::core::task::Task;
let mut backend = mock();
let task = Task::new("ship it");
backend.machine.workspaces[1].tasks = vec![task.clone()];
let prefix: String = task.id.to_string().chars().take(8).collect();
run_cli(
&["tty7", "task", "done", &prefix],
&Context::default(),
&mut backend,
);
let Some(ControlRequest::TaskPut {
workspace,
task: put,
}) = backend.control_calls.last()
else {
panic!("expected a TaskPut, got {:?}", backend.control_calls);
};
assert_eq!(*workspace, backend.machine.workspaces[1].id);
assert_eq!(put.id, task.id);
assert!(put.done.is_some());
let err = execute(
cli(&["tty7", "task", "rm", "abc"]),
&Context::default(),
&mut backend,
)
.unwrap_err();
assert!(err.to_string().contains("four"), "{err}");
}
#[test]
fn ls_and_ws_ls_are_the_same_request() {
let ctx = Context::default();
+2
View File
@@ -32,6 +32,7 @@ pub fn two_workspace_machine() -> Machine {
active_tab: None,
attachment: None,
groups: Default::default(),
tasks: Vec::new(),
};
let web = Workspace {
id: WorkspaceId::new(),
@@ -48,6 +49,7 @@ pub fn two_workspace_machine() -> Machine {
active_tab: None,
attachment: None,
groups: Default::default(),
tasks: Vec::new(),
};
let record = |id: u64, cwd: &str| PaneRecord {
id,
+116 -1
View File
@@ -9,6 +9,7 @@ use serde::{Deserialize, Serialize};
use crate::core::cli_agent::CLIAgent;
use crate::core::group_key::{AutoKey, GroupId, WorkspaceGroups};
use crate::core::session::WorkspaceId;
use crate::core::task::{Task, TaskId};
use crate::daemon::protocol::{NativeSshSpec, ShellSpec};
pub const MACHINE_FILE: &str = "machine.json";
@@ -140,6 +141,9 @@ pub struct Workspace {
/// the workspace draws the same groups in the same order.
#[serde(default, skip_serializing_if = "WorkspaceGroups::is_empty")]
pub groups: WorkspaceGroups,
/// The workspace's board. See [`crate::core::task`].
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub tasks: Vec<Task>,
}
impl Default for Workspace {
@@ -152,6 +156,7 @@ impl Default for Workspace {
active_tab: None,
attachment: None,
groups: WorkspaceGroups::default(),
tasks: Vec::new(),
}
}
}
@@ -469,6 +474,15 @@ pub enum LayoutDelta {
GroupsChanged {
groups: WorkspaceGroups,
},
/// A task was added or changed; the whole of it, which replaces any task
/// with the same id. Only sent to a peer that said it reads tasks
/// ([`LayoutDelta::is_task`]).
TaskPut {
task: Task,
},
TaskRemoved {
task: TaskId,
},
TabRestructured {
tab: Tab,
pane: Option<PaneRecord>,
@@ -483,6 +497,18 @@ pub enum LayoutDelta {
},
}
impl LayoutDelta {
/// Whether this is one of the board's deltas, which a peer that predates
/// the board cannot decode — and a control event it cannot decode takes
/// the whole link down with it.
pub fn is_task(&self) -> bool {
matches!(
self,
LayoutDelta::TaskPut { .. } | LayoutDelta::TaskRemoved { .. }
)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct SubscriberId(pub u64);
@@ -889,6 +915,53 @@ impl MachineStore {
})
}
/// Adds `task` to the workspace's board, or replaces the one with its id.
///
/// Whole-task last-writer-wins, like [`Self::workspace_set_groups`]: every
/// edit a window makes to a card is "this is the card now", and two windows
/// racing on the same card are each giving an answer the other would take.
pub fn task_put(
&self,
workspace: WorkspaceId,
task: Task,
origin: Option<SubscriberId>,
) -> io::Result<()> {
task.check().map_err(refuse)?;
self.mutate(origin, |m| {
let ws = find_workspace(m, workspace)?;
match ws.tasks.iter().position(|t| t.id == task.id) {
Some(i) => ws.tasks[i] = task.clone(),
None if ws.tasks.len() >= crate::core::task::MAX_TASKS => {
return Err(refuse(format!(
"workspace {workspace} already keeps {} tasks",
crate::core::task::MAX_TASKS
)));
}
None => ws.tasks.push(task.clone()),
}
Ok(((), vec![(workspace, LayoutDelta::TaskPut { task })]))
})
}
pub fn task_remove(
&self,
workspace: WorkspaceId,
task: TaskId,
origin: Option<SubscriberId>,
) -> io::Result<()> {
self.mutate(origin, |m| {
let ws = find_workspace(m, workspace)?;
let before = ws.tasks.len();
ws.tasks.retain(|t| t.id != task);
if ws.tasks.len() == before {
return Err(not_found(format!(
"workspace {workspace} has no task {task}"
)));
}
Ok(((), vec![(workspace, LayoutDelta::TaskRemoved { task })]))
})
}
pub fn pane_split(
&self,
workspace: WorkspaceId,
@@ -1823,7 +1896,7 @@ fn env_dir(key: &str) -> Option<PathBuf> {
.map(PathBuf::from)
}
fn unix_now() -> u64 {
pub fn unix_now() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
@@ -1936,6 +2009,48 @@ mod tests {
);
}
#[test]
fn a_workspace_keeps_its_board_across_a_restart() {
let (store, dir) = store();
let ws = store.workspace_create(None, None, None).unwrap();
let mut task = crate::core::task::Task::new("fix the flaky test");
task.prompt = "it fails one run in ten".into();
store.task_put(ws.id, task.clone(), None).unwrap();
task.title = "fix the flaky restore test".into();
store.task_put(ws.id, task.clone(), None).unwrap();
let other = crate::core::task::Task::new("second");
store.task_put(ws.id, other.clone(), None).unwrap();
store.task_remove(ws.id, other.id, None).unwrap();
let reopened = MachineStore::open(dir.path().join(MACHINE_FILE));
let back = reopened.workspace(ws.id).unwrap();
assert_eq!(
back.tasks,
vec![task],
"a put replaces by id; a remove drops it"
);
}
#[test]
fn the_board_refuses_what_it_cannot_keep() {
let (store, _dir) = store();
let ws = store.workspace_create(None, None, None).unwrap();
let untitled = crate::core::task::Task::new(" ");
assert_eq!(
store.task_put(ws.id, untitled, None).unwrap_err().kind(),
io::ErrorKind::InvalidInput
);
assert_eq!(
store
.task_remove(ws.id, crate::core::task::TaskId::new(), None)
.unwrap_err()
.kind(),
io::ErrorKind::NotFound
);
assert!(store.workspace(ws.id).unwrap().tasks.is_empty());
}
#[test]
fn the_tree_round_trips_through_the_file() {
let (store, dir) = store();
+1
View File
@@ -23,6 +23,7 @@ pub mod shells;
#[allow(dead_code)]
pub mod ssh_profile;
pub mod tab_view;
pub mod task;
pub mod term_modes;
pub mod threads;
pub mod window_state;
+440
View File
@@ -0,0 +1,440 @@
//! The board: work handed to coding agents, and where each piece of it stands.
//!
//! A [`Task`] is what the user asked for — a title, a prompt, where to run it —
//! plus the [`Run`]s that have been started for it. It lives on the workspace,
//! beside the tabs and groups, so it is kept by the same machine tree, moves
//! over the same deltas and dies with the workspace that holds it.
//!
//! What a task does *not* store is its column. Where a card sits is read off
//! the agents running it every time the board is drawn ([`column`]): a turn
//! starting, a permission prompt, a turn finishing are all things the agent
//! already reports, and a column written down beside them would only ever be
//! a second, staler copy of that. The one thing only a person can say — "this
//! is finished" — is the only verdict kept ([`Task::done`]), and even that
//! gives way when the agent is put back to work.
use serde::{Deserialize, Serialize};
use crate::core::cli_agent::{AgentSessionState, AgentStatus, CLIAgent};
use crate::core::group_key::GroupId;
use crate::core::machine::TabId;
/// How many tasks one workspace keeps. The board is a working surface rather
/// than an archive, and every one of these rides in the machine tree, which
/// is rewritten whole on every mutation.
pub const MAX_TASKS: usize = 512;
/// How long a title may be, in characters. A card shows two lines of it.
pub const MAX_TITLE_CHARS: usize = 200;
/// How long a prompt may be, in bytes. It is typed onto a command line, and
/// the whole task travels in every delta that touches it.
pub const MAX_PROMPT_BYTES: usize = 16 * 1024;
/// How many runs one task keeps. Each is a pane someone started for it; past
/// this the oldest are dropped, since a finished run is only kept to be
/// resumed and nobody resumes the tenth attempt at the same thing.
pub const MAX_RUNS: usize = 8;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(transparent)]
pub struct TaskId(uuid::Uuid);
impl TaskId {
pub fn new() -> Self {
Self(uuid::Uuid::new_v4())
}
pub fn parse(text: &str) -> Option<Self> {
uuid::Uuid::parse_str(text).ok().map(Self)
}
}
impl Default for TaskId {
fn default() -> Self {
Self::new()
}
}
impl std::fmt::Display for TaskId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.0.fmt(f)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Task {
pub id: TaskId,
pub title: String,
/// What the agent is told. Empty means the title is the whole ask.
#[serde(default, skip_serializing_if = "String::is_empty")]
pub prompt: String,
/// Where a run starts, on the workspace's own host. `None` for a task
/// written down before anyone said where it belongs.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cwd: Option<String>,
/// Which agent a run starts by default.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub agent: Option<CLIAgent>,
/// A pinned sidebar group the card is filed under. `None` leaves it to the
/// same automatic grouping a tab gets, worked out from [`Self::cwd`].
#[serde(default, skip_serializing_if = "Option::is_none")]
pub group: Option<GroupId>,
/// Unix seconds.
#[serde(default)]
pub created: u64,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub runs: Vec<Run>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub done: Option<Done>,
}
/// One agent started for a task.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Run {
pub agent: CLIAgent,
/// The tab it was opened in. A tab rather than a pane: the tab exists the
/// moment a run starts, while a pane on a remote machine has no id until
/// it lands. A tab that has gone leaves the run behind as history,
/// resumable by [`Self::session_id`].
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tab: Option<TabId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub session_id: Option<String>,
/// Unix seconds.
#[serde(default)]
pub started: u64,
}
/// The user said the task is finished.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct Done {
/// Unix seconds.
pub at: u64,
/// How many turns its live runs had finished between them when it was
/// marked. A later turn means someone put an agent back on it, and a card
/// that stayed in Done while its agent worked would be lying.
#[serde(default)]
pub turns: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum Column {
/// Nothing has started working on it.
Queued,
/// An agent is in the middle of a turn.
Running,
/// An agent is stopped on a question or a permission prompt.
NeedsInput,
/// Every agent on it has stopped, and nobody has said it is finished.
Review,
Done,
}
impl Column {
pub const ALL: [Column; 5] = [
Column::Queued,
Column::Running,
Column::NeedsInput,
Column::Review,
Column::Done,
];
}
/// What the board needs to know about the agent a run's tab is running.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Live {
pub status: AgentStatus,
/// Turns finished, as [`AgentSessionState::turns`].
pub turns: u64,
}
impl From<&AgentSessionState> for Live {
fn from(state: &AgentSessionState) -> Live {
Live {
status: state.status,
turns: state.turns,
}
}
}
/// Where one live agent session puts its card.
///
/// An agent that is open but has never been given anything to do is Queued,
/// not Running: nothing is happening, and the one thing that would move it on
/// is someone typing a prompt — which is what Queued means.
pub fn live_column(state: Live) -> Column {
match state.status {
AgentStatus::Waiting => Column::NeedsInput,
AgentStatus::Working => Column::Running,
AgentStatus::Done => Column::Review,
AgentStatus::Idle if state.turns > 0 => Column::Review,
AgentStatus::Idle => Column::Queued,
}
}
/// Where a task's card sits, given what the agents in its runs' tabs are
/// doing now.
///
/// `live` answers for a run whose tab is still open and running an agent; a
/// run it has no answer for is taken to be over. The most urgent run wins: one
/// agent asking a question is the card's news even while another is busy,
/// and any agent still busy outranks the user's say-so that it is finished.
pub fn column(task: &Task, live: impl Fn(&Run) -> Option<Live>) -> Column {
let states: Vec<Live> = task.runs.iter().filter_map(live).collect();
if states.iter().any(|s| s.status == AgentStatus::Waiting) {
return Column::NeedsInput;
}
if states.iter().any(|s| s.status == AgentStatus::Working) {
return Column::Running;
}
if let Some(done) = task.done
&& turns(&states) <= done.turns
{
return Column::Done;
}
if task.runs.is_empty() {
return Column::Queued;
}
// An agent open on a run that has not taken a turn yet was just started
// for this task and is picking its prompt up.
if states.iter().any(|&s| live_column(s) == Column::Queued) {
return Column::Running;
}
Column::Review
}
/// The turns a task's live runs have finished between them, which is what a
/// [`Done`] mark is measured against.
pub fn live_turns(task: &Task, live: impl Fn(&Run) -> Option<Live>) -> u64 {
let states: Vec<Live> = task.runs.iter().filter_map(live).collect();
turns(&states)
}
fn turns(states: &[Live]) -> u64 {
states
.iter()
.fold(0u64, |sum, s| sum.saturating_add(s.turns))
}
impl Task {
pub fn new(title: impl Into<String>) -> Task {
Task {
id: TaskId::new(),
title: title.into(),
prompt: String::new(),
cwd: None,
agent: None,
group: None,
created: crate::core::machine::unix_now(),
runs: Vec::new(),
done: None,
}
}
/// What the agent is told when a run starts: the prompt, or the title when
/// there is none.
pub fn ask(&self) -> &str {
let prompt = self.prompt.trim();
if prompt.is_empty() {
self.title.trim()
} else {
prompt
}
}
/// Whether any run of this task was opened in `tab`.
pub fn runs_in(&self, tab: TabId) -> bool {
self.runs.iter().any(|r| r.tab == Some(tab))
}
/// Records a run, dropping the oldest past [`MAX_RUNS`].
pub fn push_run(&mut self, run: Run) {
self.runs.push(run);
if self.runs.len() > MAX_RUNS {
let excess = self.runs.len() - MAX_RUNS;
self.runs.drain(..excess);
}
}
/// Whether the task is fit to keep, and if not, why — in words for
/// whoever sent it.
pub fn check(&self) -> Result<(), String> {
if self.title.trim().is_empty() {
return Err("a task needs a title".into());
}
if self.title.chars().count() > MAX_TITLE_CHARS {
return Err(format!(
"a task title is at most {MAX_TITLE_CHARS} characters"
));
}
if self.prompt.len() > MAX_PROMPT_BYTES {
return Err(format!(
"a task prompt is at most {} KiB",
MAX_PROMPT_BYTES / 1024
));
}
if self.runs.len() > MAX_RUNS {
return Err(format!("a task keeps at most {MAX_RUNS} runs"));
}
Ok(())
}
}
/// The argument that hands `agent` its first prompt while leaving it open for
/// the conversation that follows, or `None` for an agent whose command line
/// cannot do that — one that takes a prompt only to answer it and exit, or
/// takes none at all.
///
/// The flag, when there is one, comes before the prompt: `gemini -i "…"`.
pub fn prompt_flag(agent: CLIAgent) -> Option<Option<&'static str>> {
match agent {
CLIAgent::Claude | CLIAgent::Codex | CLIAgent::Cursor => Some(None),
CLIAgent::Gemini | CLIAgent::Qwen | CLIAgent::Copilot => Some(Some("-i")),
CLIAgent::OpenCode => Some(Some("--prompt")),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn live(status: AgentStatus, turns: u64) -> Live {
Live { status, turns }
}
/// A task with one run per entry, each in a tab of its own.
fn task_with(runs: usize) -> (Task, Vec<TabId>) {
let mut t = Task::new("fix it");
let tabs: Vec<TabId> = (0..runs).map(|_| TabId::new()).collect();
for &tab in &tabs {
t.push_run(Run {
agent: CLIAgent::Claude,
tab: Some(tab),
session_id: None,
started: 0,
});
}
(t, tabs)
}
fn col(task: &Task, states: &[(TabId, Live)]) -> Column {
column(task, |r| {
states
.iter()
.find(|(tab, _)| Some(*tab) == r.tab)
.map(|(_, s)| *s)
})
}
#[test]
fn a_task_nobody_started_is_queued() {
assert_eq!(col(&task_with(0).0, &[]), Column::Queued);
}
#[test]
fn each_live_status_has_its_column() {
let (t, tabs) = task_with(1);
let at = |l| [(tabs[0], l)];
assert_eq!(col(&t, &at(live(AgentStatus::Working, 0))), Column::Running);
assert_eq!(
col(&t, &at(live(AgentStatus::Waiting, 0))),
Column::NeedsInput
);
assert_eq!(col(&t, &at(live(AgentStatus::Done, 1))), Column::Review);
assert_eq!(col(&t, &at(live(AgentStatus::Idle, 2))), Column::Review);
}
#[test]
fn a_run_that_has_not_picked_up_its_prompt_yet_counts_as_running() {
let (t, tabs) = task_with(1);
assert_eq!(
col(&t, &[(tabs[0], live(AgentStatus::Idle, 0))]),
Column::Running
);
}
#[test]
fn a_run_whose_tab_is_gone_leaves_the_task_to_review() {
assert_eq!(col(&task_with(1).0, &[]), Column::Review);
}
#[test]
fn the_most_urgent_run_wins() {
let (t, tabs) = task_with(2);
let states = [
(tabs[0], live(AgentStatus::Working, 0)),
(tabs[1], live(AgentStatus::Waiting, 0)),
];
assert_eq!(col(&t, &states), Column::NeedsInput);
let states = [
(tabs[0], live(AgentStatus::Working, 0)),
(tabs[1], live(AgentStatus::Done, 1)),
];
assert_eq!(col(&t, &states), Column::Running);
}
#[test]
fn done_holds_until_an_agent_takes_another_turn() {
let (mut t, tabs) = task_with(1);
t.done = Some(Done { at: 1, turns: 1 });
let at = |l| [(tabs[0], l)];
assert_eq!(col(&t, &at(live(AgentStatus::Done, 1))), Column::Done);
assert_eq!(col(&t, &[]), Column::Done);
assert_eq!(col(&t, &at(live(AgentStatus::Working, 1))), Column::Running);
assert_eq!(col(&t, &at(live(AgentStatus::Done, 2))), Column::Review);
}
#[test]
fn a_task_marked_done_before_it_ever_ran_is_done() {
let (mut t, _) = task_with(0);
t.done = Some(Done { at: 1, turns: 0 });
assert_eq!(col(&t, &[]), Column::Done);
}
#[test]
fn a_live_agent_nobody_has_asked_anything_is_queued() {
assert_eq!(live_column(live(AgentStatus::Idle, 0)), Column::Queued);
assert_eq!(live_column(live(AgentStatus::Idle, 1)), Column::Review);
}
#[test]
fn runs_past_the_cap_drop_the_oldest() {
let (t, tabs) = task_with(MAX_RUNS + 3);
assert_eq!(t.runs.len(), MAX_RUNS);
assert_eq!(t.runs[0].tab, Some(tabs[3]));
}
#[test]
fn check_refuses_what_the_tree_should_not_keep() {
assert!(Task::new(" ").check().is_err());
assert!(Task::new("x".repeat(MAX_TITLE_CHARS + 1)).check().is_err());
let mut t = Task::new("ok");
t.prompt = "p".repeat(MAX_PROMPT_BYTES + 1);
assert!(t.check().is_err());
assert!(Task::new("ok").check().is_ok());
}
#[test]
fn the_ask_falls_back_to_the_title() {
let mut t = Task::new(" Fix the flaky test ");
assert_eq!(t.ask(), "Fix the flaky test");
t.prompt = " details\n".into();
assert_eq!(t.ask(), "details");
}
#[test]
fn a_task_round_trips_and_an_old_one_reads() {
let (mut t, _) = task_with(1);
t.prompt = "go".into();
t.done = Some(Done { at: 9, turns: 2 });
let json = serde_json::to_string(&t).unwrap();
assert_eq!(serde_json::from_str::<Task>(&json).unwrap(), t);
let bare: Task =
serde_json::from_str(r#"{"id":"6f1c0b52-6a7e-4d53-9d1e-0a1b2c3d4e5f","title":"t"}"#)
.unwrap();
assert!(bare.runs.is_empty() && bare.done.is_none() && bare.prompt.is_empty());
}
}
+26
View File
@@ -158,6 +158,12 @@ pub mod feature {
/// request, and sending it would take the link down, so a client asks for
/// the name first and tells the user the server needs updating instead.
pub const CONTENT_SEARCH: &str = "content-search";
/// Both ways. From a server: it keeps a board on each workspace and
/// answers [`ControlRequest::TaskPut`] and [`ControlRequest::TaskRemove`].
/// From a client (in [`ControlHello::features`]): it can decode the
/// board's layout deltas, which a server holds back from a client that
/// never said so rather than send it an event it would choke on.
pub const TASKS: &str = "tasks";
}
pub use crate::host::{
@@ -341,6 +347,17 @@ pub enum ControlRequest {
tab: TabId,
hibernated: bool,
},
/// Add a task to the workspace's board, or replace the one with its id.
/// Gated on [`feature::TASKS`].
TaskPut {
workspace: WorkspaceId,
task: crate::core::task::Task,
},
/// Gated on [`feature::TASKS`].
TaskRemove {
workspace: WorkspaceId,
task: crate::core::task::TaskId,
},
PaneSplit {
workspace: WorkspaceId,
pane: u64,
@@ -453,6 +470,8 @@ impl ControlRequest {
| TabSetGroup { .. }
| WorkspaceSetGroups { .. }
| TabSetHibernated { .. }
| TaskPut { .. }
| TaskRemove { .. }
| PaneSplit { .. }
| PaneClose { .. }
| PaneSetRatio { .. }
@@ -667,6 +686,11 @@ pub struct ControlHello {
pub client_hostname: String,
#[serde(default)]
pub gui: bool,
/// What this client can take that an older one could not — the client's
/// half of [`ControlHelloOk::features`]. A server predating the field
/// ignores it, and a client predating it is taken to read nothing new.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub features: Vec<String>,
}
impl ControlHello {
@@ -677,6 +701,7 @@ impl ControlHello {
client_token: client_token.into(),
client_hostname: client_hostname.into(),
gui: false,
features: vec![feature::TASKS.to_string()],
}
}
@@ -1794,6 +1819,7 @@ mod tests {
client_token: "tok".into(),
client_hostname: "laptop".into(),
gui: false,
features: Vec::new(),
}
}
+79 -2
View File
@@ -282,7 +282,8 @@ where
}
};
let machine_sub = subscribe_machine(&services, &sink);
let reads_tasks = hello.features.iter().any(|f| f == feature::TASKS);
let machine_sub = subscribe_machine(&services, &sink, reads_tasks);
let conn = Arc::new(Conn {
host,
@@ -390,6 +391,7 @@ fn handshake<R: Read>(
}
if services.machine.is_some() {
features.push(feature::MACHINE_TREE.to_string());
features.push(feature::TASKS.to_string());
}
// A sleeping tab is a mark in the tree *and* stopped panes; a peer that can
// do only half of that would leave a tab asleep with its shells running,
@@ -825,6 +827,16 @@ fn run_request(
.workspace_set_groups(workspace, groups, conn.machine_origin)?;
(ReplyOk::Unit, Vec::new())
}
ControlRequest::TaskPut { workspace, task } => {
conn.machine()?
.task_put(workspace, task, conn.machine_origin)?;
(ReplyOk::Unit, Vec::new())
}
ControlRequest::TaskRemove { workspace, task } => {
conn.machine()?
.task_remove(workspace, task, conn.machine_origin)?;
(ReplyOk::Unit, Vec::new())
}
ControlRequest::TabSetHibernated {
workspace,
tab,
@@ -1212,13 +1224,23 @@ impl Conn {
}
}
fn subscribe_machine(services: &Services, sink: &Arc<Sink>) -> Option<machine::Subscription> {
/// `reads_tasks` is whether the peer said it can decode the board's deltas. A
/// peer that did not is simply not told about the board: it has nowhere to
/// show one, and an event it cannot decode would end the link.
fn subscribe_machine(
services: &Services,
sink: &Arc<Sink>,
reads_tasks: bool,
) -> Option<machine::Subscription> {
let store = services.machine.as_ref()?;
let (tx, rx) = smol::channel::bounded::<(String, machine::LayoutDelta)>(LAYOUT_EVENT_QUEUE);
let lagged = Arc::new(std::sync::atomic::AtomicBool::new(false));
let saw_drop = Arc::clone(&lagged);
let subscription = store.subscribe(Arc::new(
move |workspace: &str, delta: &machine::LayoutDelta| {
if !reads_tasks && delta.is_task() {
return;
}
if tx.try_send((workspace.to_string(), delta.clone())).is_err() {
saw_drop.store(true, Ordering::Release);
log::warn!(
@@ -2360,6 +2382,7 @@ mod tests {
client_token: token.to_string(),
client_hostname: hostname.to_string(),
gui: false,
features: vec![feature::TASKS.to_string()],
}
}
@@ -3514,6 +3537,60 @@ mod tests {
}
}
/// A peer that predates the board would fail to decode a task delta, and a
/// control event it cannot decode ends its link. So only a peer that said
/// it reads them is sent them; the older one sees the rest of the tree
/// move as before.
#[test]
fn board_deltas_reach_only_the_peers_that_read_them() {
let (services, _dir) = workspace_services();
let store = Arc::clone(services.machine.as_ref().unwrap());
let ws = store.workspace_create(None, None, None).unwrap();
let ((mut old, old_ok), _s1) = raw_hello(
services.clone(),
ControlHello {
features: Vec::new(),
..ControlHello::host_rpc("old", "old")
},
);
let ((mut new, _), _s2) = raw_hello(services.clone(), ControlHello::host_rpc("new", "new"));
assert!(old_ok.has_feature(feature::TASKS));
// A request answered means the connection is past its subscription.
round_trip(&mut old, 1, ControlRequest::Ping);
round_trip(&mut new, 1, ControlRequest::Ping);
let task = crate::core::task::Task::new("ship it");
store.task_put(ws.id, task.clone(), None).unwrap();
store
.workspace_rename(ws.id, Some("after".into()), None)
.unwrap();
let until_rename = |sock: &mut UnixStream| {
sock.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
let mut seen = Vec::new();
loop {
match ControlServerMsg::read(sock).unwrap() {
ControlServerMsg::Event(ControlEvent::Layout { delta, .. }) => {
let done = matches!(delta, machine::LayoutDelta::WorkspaceRenamed { .. });
seen.push(delta);
if done {
return seen;
}
}
_ => continue,
}
}
};
assert!(
!until_rename(&mut old).iter().any(|d| d.is_task()),
"a peer that never said it reads tasks is not sent one"
);
assert!(
until_rename(&mut new).contains(&machine::LayoutDelta::TaskPut { task }),
"a peer that did gets the card"
);
}
#[test]
fn a_second_client_takes_the_workspace_and_the_first_is_told() {
let (services, _dir) = workspace_services();
+15
View File
@@ -285,6 +285,21 @@ Four rules from it survive even if you read nothing else:
- **After the first send into a new pane, confirm the command left the
prompt** — the swallowed-Enter check above.
### The board
Each workspace has a board: tasks for agents, in five columns (`queued`,
`running`, `needs-input`, `review`, `done`) worked out from what each task's
agent is doing. When the user wants work written down rather than started now,
put it there; they start it from the GUI.
```bash
tty7 task add "Fix the flaky restore test" --prompt-file spec.md --cwd ~/src/api --agent claude
tty7 task ls # this workspace's tasks and their columns (--json for runs, sessions)
tty7 task done 3f2a9c1e # any unique prefix of the id, at least four characters
tty7 task reopen 3f2a9c1e
tty7 task rm 3f2a9c1e
```
## Looking around
```bash