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 Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

9 changes: 9 additions & 0 deletions PLAN.md
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,15 @@ One package is one branch and one pull request. A package that adds a mechanism
| 14 | The surface of Appendix A | 5, 7 | Appendix A | `get` by anchor and by version, with `related` and its weights; `search` with filters and snippets; `recall` with `context_tokens` and `answer_kind`; `approve`, `erase`, `export` and `declassify` as memory tools | Act 1 |
| 15 | Long timescales | 2, 7 | §7 Keeping strategy in touch with the world; §9 outage; §12; Appendix E Recovery | The identity form `human:<authority>,<date>:<id>`; an ownership registry that flags a domain without an owner; personal domains; recovery bundles with a claim epoch, and a restore drill; work from T0 during an outage; a scheduled task that gathers evidence for the horizon scanner, who is a person | Act 10: a restore drill |

Landed so far:

| # | Pull request | What is still open in the package |
|---|---|---|
| 0 | #69 | Nothing |
| 1 | #71 | Nothing |
| 2 | #73 | An attempt key on every operation; an uncertain external effect blocking reassignment; waiting on an outside event as a state; the record's remaining fields (`parent`, `scores`, `implements`, `reversibility`, `artifacts_touched`, `policy_context`) |
| 3 | the pull request after #73 | A suite that gates a version of a definition; the data scope enforced; a definition required by default, once the system's own filings name one; tools offered a few at boot and the rest by search |

What stays unbuilt after these, and stays on README's list: federation across domains, authentication of people, logprob-based drift, and accelerator time, which only a gateway can count. So do the items on that list that need another machine or an outside service more than they need code: the limits on re-enactment, harness settings and decomposition templates in the platform domain, bytes stored and resource limits on the sandbox, the authoring of agent definitions, the search tool against its live provider, a harness under Seatbelt on a Mac, and the Gofannon bridge against a live deployment, whose agents make model calls that the egress does not meter.

## Toward the production shape
Expand Down
19 changes: 11 additions & 8 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -108,14 +108,17 @@ two weeks. It is also the reason that none of it is the production shape.
- accelerator time, which only the gateway can count
- bytes stored
- CPU or memory limits on the sandbox
- **The task record, in part.** A claim has no generation. The ledger accepts a caller
when the caller's name is the name of the holder and the lease of the task has not
expired. It checks nothing else, and two sessions of one harness have the same name. A task has no `blocked` state: the ledger computes a block when a session
claims. The ledger does not record a refused claim. An uncertain external effect does
not stop a new claim.
- **Task definitions and autonomy levels.** A task does not name a task definition, and
triage admits a task without one. A session gets its tools from the grants file, not
from a definition. There is no autonomy level, no held-out suite and no evaluator.
- **The task record, in part.** A claim has a generation, and the ledger refuses a
holder that gives an earlier one. A task can be `blocked` on a task that it depends on.
These parts are not built: an attempt key for each operation, a block on a new claim
while an external effect is uncertain, and a `blocked` state for a task that waits on an
outside event. The generation does not prove who the caller is.
- **Task definitions, in part, and autonomy levels.** A task can name a task definition.
It then takes the checks of the definition, and its holder can call only the tools that
the definition declares. Triage requires a definition only when the grants say so,
because the tasks that the system files name none. No suite gates a version of a
definition, and the data scope is not enforced. The autonomy level has no effect, and
there is no held-out suite and no evaluator.
- **Positions.** There is no `position` entry, and a task cannot be pinned.
- **Approval events.** An approval is the completion of a promotion task. It has no
scope and no expiry, and the perimeter does not check it when an action runs. There is
Expand Down
1 change: 1 addition & 0 deletions crates/client/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -590,6 +590,7 @@ mod tests {
created_by: "human:test:me".into(),
priority: 0,
classification: Default::default(),
..dm_ledger::TaskSpec::default()
}
}

Expand Down
71 changes: 69 additions & 2 deletions crates/ledger/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -155,7 +155,7 @@ impl Status {
}

/// What a caller files. Everything else on a [`Task`] the ledger fills in.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct TaskSpec {
/// The domain whose ledger partition holds it.
pub domain: String,
Expand Down Expand Up @@ -183,6 +183,33 @@ pub struct TaskSpec {
/// Higher is claimed first.
#[serde(default)]
pub priority: i64,
/// The task definition this task runs under, as `<name>@<version>`: the record in
/// the platform's domain for its kind of work.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub definition: Option<String>,
/// The checks that say this task is done, its own or those its definition gave it.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub acceptance: Vec<String>,
}

/// The fields of a task that the ledger keeps and never queries by, as one JSON column.
#[derive(Default, Serialize, Deserialize)]
struct Extra {
#[serde(default, skip_serializing_if = "Option::is_none")]
definition: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
acceptance: Vec<String>,
}

impl TaskSpec {
/// The `extra` column of this spec's row.
pub(crate) fn extra(&self) -> String {
let extra = Extra {
definition: self.definition.clone(),
acceptance: self.acceptance.clone(),
};
serde_json::to_string(&extra).unwrap_or_default()
}
}

/// A task as the ledger holds it.
Expand Down Expand Up @@ -589,7 +616,7 @@ impl RawPromotion {
const TASK_COLUMNS: &str =
"id, domain, title, body, requirements, depends_on, origin, upstream_event,
created_by, label_rank, priority, created, status, holder, lease_until, attempts, handoff,
reason, superseded_by, posted_to";
reason, superseded_by, posted_to, extra";

/// A task row as the database returns it, before the ledger has checked it.
struct RawTask {
Expand All @@ -613,6 +640,7 @@ struct RawTask {
reason: Option<String>,
superseded_by: Option<String>,
posted_to: Option<String>,
extra: Option<String>,
}

impl RawTask {
Expand All @@ -626,6 +654,14 @@ impl RawTask {
};
let requirements = json_list(&self.requirements, "requirements")?;
let depends_on = json_list(&self.depends_on, "depends_on")?;
// a row from before the column has none, and reads as a task with nothing more
let extra: Extra = match self.extra.as_deref() {
None | Some("") => Extra::default(),
Some(text) => serde_json::from_str(text).map_err(|_| Error::BadRow {
id: id.clone(),
what: format!("extra {text:?}"),
})?,
};
Ok(Task {
spec: TaskSpec {
domain: self.domain,
Expand All @@ -638,6 +674,8 @@ impl RawTask {
created_by: self.created_by,
classification: label_from_rank(&id, self.label_rank)?,
priority: self.priority,
definition: extra.definition,
acceptance: extra.acceptance,
},
created: parse_time(&id, &self.created)?,
status: Status::parse(&id, &self.status)?,
Expand Down Expand Up @@ -1786,6 +1824,7 @@ mod tests {
created_by: "human:test:me".into(),
classification: Label::Public,
priority: 0,
..TaskSpec::default()
}
}

Expand Down Expand Up @@ -2096,6 +2135,34 @@ mod tests {
assert_eq!(l.get(&rests.id).await.unwrap().status, Status::Proposed);
}

/// A ledger file from before a task carried a definition: no `extra` column.
#[tokio::test]
async fn on_sqlite_a_ledger_from_before_the_extra_column_is_given_it() {
let dir = tempfile::tempdir().unwrap();
let file = dir.path().join("ledger.db");
let id = {
let mut l = Ledger::open(&file).unwrap();
admitted(&mut l, spec("Old")).await.id
};
rusqlite::Connection::open(&file)
.unwrap()
.execute_batch("ALTER TABLE tasks DROP COLUMN extra;")
.unwrap();
let mut l = Ledger::open(&file).unwrap();
assert_eq!(l.get(&id).await.unwrap().spec.definition, None);
let mut defined = spec("New");
defined.definition = Some("review-change@mem-p-1".into());
defined.acceptance = vec!["It is checked.".into()];
let new = l.create(defined.clone()).await.unwrap().task;
assert_eq!(new.spec.definition, defined.definition);
// a rebuild reads the same out of the events
l.rebuild().await.unwrap();
assert_eq!(
l.get(&new.id).await.unwrap().spec.acceptance,
defined.acceptance
);
}

/// Every task row's state set to a name this code does not have.
async fn make_stale(b: &Site) {
match b {
Expand Down
28 changes: 25 additions & 3 deletions crates/ledger/src/postgres.rs
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,8 @@ CREATE TABLE IF NOT EXISTS {s}.tasks (
handoff TEXT,
reason TEXT,
superseded_by TEXT,
posted_to TEXT
posted_to TEXT,
extra TEXT
);
CREATE INDEX IF NOT EXISTS tasks_queue ON {s}.tasks (status, priority DESC, created);
CREATE UNIQUE INDEX IF NOT EXISTS tasks_origin ON {s}.tasks (origin, upstream_event)
Expand Down Expand Up @@ -176,6 +177,7 @@ fn task_from_row(row: &Row) -> Result<Task, Error> {
reason: row.try_get(17)?,
superseded_by: row.try_get(18)?,
posted_to: row.try_get(19)?,
extra: row.try_get(20)?,
}
.into_task()
}
Expand Down Expand Up @@ -213,6 +215,25 @@ impl Db {
/// stands, such a task would be in no list and no queue, without a word.
async fn rebuild_stale_projections(&mut self) -> Result<(), Error> {
for (domain, s) in self.schemas().await? {
// a schema from before a task carried more than the ledger queries by: asked
// of the catalog first, so a connect takes no lock on a table that has it
let has_extra = self
.client
.query_opt(
"SELECT 1 FROM information_schema.columns
WHERE table_schema = $1 AND table_name = 'tasks' AND column_name = 'extra'",
// the catalog holds the name without the quotes the schema is written in
&[&s.trim_matches('"')],
)
.await?
.is_some();
if !has_extra {
self.client
.batch_execute(&format!(
"ALTER TABLE {s}.tasks ADD COLUMN IF NOT EXISTS extra TEXT"
))
.await?;
}
let row = self
.client
.query_one(
Expand Down Expand Up @@ -1160,8 +1181,8 @@ async fn insert_task<C: GenericClient>(
conn.execute(
&format!(
"INSERT INTO {s}.tasks (id, domain, title, body, requirements, depends_on, origin,
upstream_event, created_by, label_rank, priority, created, status)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, 'proposed')"
upstream_event, created_by, label_rank, priority, created, status, extra)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, 'proposed', $13)"
),
&[
&id,
Expand All @@ -1176,6 +1197,7 @@ async fn insert_task<C: GenericClient>(
&(spec.classification as i64),
&spec.priority,
&crate::fmt_time(at),
&spec.extra(),
],
)
.await?;
Expand Down
18 changes: 15 additions & 3 deletions crates/ledger/src/sqlite.rs
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,8 @@ CREATE TABLE IF NOT EXISTS tasks (
handoff TEXT,
reason TEXT,
superseded_by TEXT,
posted_to TEXT
posted_to TEXT,
extra TEXT
);
CREATE INDEX IF NOT EXISTS tasks_queue ON tasks (domain, status, priority DESC, created);
DROP INDEX IF EXISTS tasks_origin;
Expand Down Expand Up @@ -88,6 +89,7 @@ fn task_from_row(row: &Row<'_>) -> Result<Task, Error> {
reason: row.get(17)?,
superseded_by: row.get(18)?,
posted_to: row.get(19)?,
extra: row.get(20)?,
}
.into_task()
}
Expand Down Expand Up @@ -118,6 +120,15 @@ impl Db {
if has_posted == 0 {
conn.execute_batch("ALTER TABLE tasks ADD COLUMN posted_to TEXT;")?;
}
// and one from before a task carried more than the ledger queries by
let has_extra: i64 = conn.query_row(
"SELECT COUNT(*) FROM pragma_table_info('tasks') WHERE name = 'extra'",
[],
|r| r.get(0),
)?;
if has_extra == 0 {
conn.execute_batch("ALTER TABLE tasks ADD COLUMN extra TEXT;")?;
}
migrate_promotions(&conn)?;
// a projection whose state names this code does not have, as a ledger from before
// the states were `open` and `done`: read as it stands, its tasks would be in no
Expand Down Expand Up @@ -755,8 +766,8 @@ fn insert_task(
let depends_on = serde_json::to_string(&spec.depends_on).unwrap_or_default();
conn.execute(
"INSERT INTO tasks (id, domain, title, body, requirements, depends_on, origin,
upstream_event, created_by, label_rank, priority, created, status)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, 'proposed')",
upstream_event, created_by, label_rank, priority, created, status, extra)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, 'proposed', ?13)",
params![
id,
spec.domain,
Expand All @@ -770,6 +781,7 @@ fn insert_task(
spec.classification as i64,
spec.priority,
crate::fmt_time(at),
spec.extra(),
],
)?;
Ok(())
Expand Down
2 changes: 2 additions & 0 deletions crates/memory/src/auditor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,7 @@ pub async fn file_contradictions(
created_by: String::new(),
classification: dm_core::Label::Public,
priority: 1,
..dm_ledger::TaskSpec::default()
})
.await?;
if !created.deduplicated {
Expand Down Expand Up @@ -326,6 +327,7 @@ pub async fn file(
created_by: String::new(),
classification: dm_core::Label::Public,
priority: 0,
..dm_ledger::TaskSpec::default()
})
.await?;
if !created.deduplicated {
Expand Down
1 change: 1 addition & 0 deletions crates/memory/src/canary.rs
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,7 @@ pub async fn file(
created_by: String::new(),
classification: dm_core::Label::Public,
priority: 10,
..dm_ledger::TaskSpec::default()
})
.await?;
if !created.deduplicated {
Expand Down
1 change: 1 addition & 0 deletions crates/memory/src/curation.rs
Original file line number Diff line number Diff line change
Expand Up @@ -299,6 +299,7 @@ pub async fn file_alarms(
created_by: String::new(),
classification: dm_core::Label::Public,
priority: 5,
..dm_ledger::TaskSpec::default()
})
.await?;
if !created.deduplicated {
Expand Down
1 change: 1 addition & 0 deletions crates/memory/src/curator.rs
Original file line number Diff line number Diff line change
Expand Up @@ -443,6 +443,7 @@ pub async fn file(
created_by: String::new(),
classification: dm_core::Label::Public,
priority: -1,
..dm_ledger::TaskSpec::default()
})
.await?;
if created.deduplicated {
Expand Down
1 change: 1 addition & 0 deletions crates/memory/src/drift.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1015,6 +1015,7 @@ pub async fn file(
created_by: String::new(),
classification: dm_core::Label::Public,
priority: 5,
..dm_ledger::TaskSpec::default()
})
.await?;
Ok((!created.deduplicated).then_some(created.task.id))
Expand Down
1 change: 1 addition & 0 deletions crates/memory/src/gofannon.rs
Original file line number Diff line number Diff line change
Expand Up @@ -340,6 +340,7 @@ mod tests {
created_by: String::new(),
classification: dm_core::Label::Public,
priority: 0,
..dm_ledger::TaskSpec::default()
})
.await
.unwrap()
Expand Down
12 changes: 12 additions & 0 deletions crates/memory/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1117,6 +1117,13 @@ enum TaskOp {
depends_on: Vec<String>,
#[arg(long, default_value_t = 0)]
priority: i64,
/// The task definition this kind of work runs under, by name.
#[arg(long)]
definition: Option<String>,
/// A check that says this task is done; repeat for more. Without one, the task
/// takes those of its definition.
#[arg(long = "accept")]
acceptance: Vec<String>,
},
/// Admit a proposed task.
Admit { id: String },
Expand Down Expand Up @@ -4715,6 +4722,7 @@ async fn policy_lint(m: &Memory, file: bool) -> Fallible {
created_by: String::new(),
classification: dm_core::Label::Internal,
priority: 4,
..dm_ledger::TaskSpec::default()
})
.await?;
if !created.deduplicated {
Expand Down Expand Up @@ -4909,6 +4917,8 @@ async fn task(m: &Memory, op: TaskOp) -> Fallible {
requirements,
depends_on,
priority,
definition,
acceptance,
} => {
let c = m
.create_task(TaskSpec {
Expand All @@ -4922,6 +4932,8 @@ async fn task(m: &Memory, op: TaskOp) -> Fallible {
created_by: String::new(),
classification: Label::Public,
priority,
definition,
acceptance,
})
.await?;
json(&c.task)
Expand Down
Loading
Loading