diff --git a/docs/RUST_PROTOTYPE.md b/docs/RUST_PROTOTYPE.md index dd31ea4ee..626a7b4b9 100644 --- a/docs/RUST_PROTOTYPE.md +++ b/docs/RUST_PROTOTYPE.md @@ -798,6 +798,26 @@ counted as a binding drop; only `prefix_binding_mismatch` is. A live refusal was not reproduced, so the fallback path is covered by parser and synthetic-endpoint tests. +A turn that ends with `provider_refusal` leaves what the declined request +added since the model last answered (the prompt, when it was the turn's first +request, or the latest tool results and any steer) out of later requests, as +Anthropic's [streaming refusals](https://platform.claude.com/docs/en/test-and-evaluate/strengthen-guardrails/handle-streaming-refusals) +guidance says to: "you must reset the conversation context before +continuing... Attempting to continue without resetting will result in +continued refusals." Each item becomes a note naming the refusal's category, +so the model knows something was declined: a tool result keeps its call with +`[This tool result is left out: ...]`, and a message becomes `[A message here +is left out: ...]`. A note is padded with spaces to the size of the item it +replaces, so the context's byte accounting does not change; a shorter wording +is used where the full one does not fit, and an item too small for any keeps +its own. Everything before it is unchanged, so no thinking binding or prompt +cache is touched. The originals stay in the `declined` table (schema 61): +turn listings and the history reads (`history_items`, `history_read`, +`history_search`, `result`) show them, and a retry with the same request id +still matches the person's prompt. A node another bot also reaches, as a fork taken while the +request was out does, is that bot's history too: it and what came before it +stay as they are. + `effort` (`low`, `medium`, `high`, `xhigh`, `max`) maps to Responses `reasoning.effort` with summaries requested, and to Anthropic adaptive thinking (`thinking.type: adaptive` with summarized display) plus `output_config.effort`. diff --git a/src/store/context.rs b/src/store/context.rs index 0bab33c9e..f55b7c283 100644 --- a/src/store/context.rs +++ b/src/store/context.rs @@ -516,6 +516,58 @@ pub fn model_output(item: &[u8]) -> bool { }) } +/// What a request a safety classifier declined added since the model last +/// answered goes to later requests as this, in its place: a tool result +/// keeps its call with a line saying it was left out, and a message says +/// the same, the person's first fully and any later one briefly. Each is +/// padded with spaces to `len` bytes, the bytes it replaces, so nothing +/// counted for the context changes: the longest wording that fits, or none +/// when not even the shortest does. `category` is the refusal's, if named. +pub fn declined( + family: Family, + item: &[u8], + category: Option<&str>, + first: bool, + len: usize, +) -> Result>> { + let why = match category { + Some(category) => format!( + "the provider's safety classifier declined the request that carried it (category: {category})" + ), + None => "the provider's safety classifier declined the request that carried it".to_owned(), + }; + let value: serde_json::Value = serde_json::from_slice(item)?; + let kind = value["type"].as_str().unwrap_or_default(); + let calls: Vec<_> = (value["content"].as_array().into_iter().flatten()) + .filter(|block| block["type"] == "tool_result") + .map(|block| block["tool_use_id"].clone()) + .collect(); + let result = RESULT_KINDS.contains(&kind) || !calls.is_empty(); + let full = match (result, first) { + (true, _) => format!("[This tool result is left out: {why}.]"), + (false, true) => format!("[A message here is left out: {why}.]"), + (false, false) => "[Another message here is left out, for the same reason.]".to_owned(), + }; + let short = "[Left out: the provider's safety classifier declined it.]"; + for text in [full.as_str(), short, "[Left out.]"] { + let mut note = if RESULT_KINDS.contains(&kind) { + serde_json::to_vec(&json!({"type":kind,"call_id":value["call_id"],"output":text}))? + } else if result { + let content: Vec<_> = (calls.iter()) + .map(|id| json!({"type":"tool_result","tool_use_id":id,"content":text})) + .collect(); + serde_json::to_vec(&json!({"role":"user","content":content}))? + } else { + family.user_item(text)? + }; + if note.len() <= len { + note.resize(len, b' '); + return Ok(Some(note)); + } + } + Ok(None) +} + /// Stable pinned blocks retain their own Anthropic cache breakpoints. pub fn pinned_item(family: Family, text: &str) -> Result> { match family { @@ -530,6 +582,94 @@ pub fn pinned_item(family: Family, text: &str) -> Result> { mod tests { use super::*; + #[test] + fn a_declined_request_s_additions_become_notes_of_their_own_length() { + let read = |item: Option>| { + let item = item.unwrap(); + ( + item.len(), + serde_json::from_slice::(&item).unwrap(), + ) + }; + let why = "the provider's safety classifier declined the request that carried it (category: cyber)"; + let output = "secret ".repeat(40); + let result = Family::Anthropic + .tool_result_item("toolu_1", false, &output) + .unwrap(); + assert_eq!( + read( + declined( + Family::Anthropic, + &result, + Some("cyber"), + false, + result.len() + ) + .unwrap() + ), + ( + result.len(), + json!({"role":"user","content":[{"type":"tool_result","tool_use_id":"toolu_1", + "content":format!("[This tool result is left out: {why}.]")}]}) + ) + ); + let custom = Family::Responses + .tool_result_item("call_1", true, &output) + .unwrap(); + assert_eq!( + read(declined(Family::Responses, &custom, None, true, custom.len()).unwrap()).1, + json!({"type":"custom_tool_call_output","call_id":"call_1","output": + "[This tool result is left out: the provider's safety classifier declined the request that carried it.]"}) + ); + // The person's first message says so in full, a later one briefly. + let prompt = Family::Responses.user_item(&output).unwrap(); + assert_eq!( + read( + declined( + Family::Responses, + &prompt, + Some("cyber"), + true, + prompt.len() + ) + .unwrap() + ) + .1, + json!({"role":"user","content":[{"type":"input_text", + "text":format!("[A message here is left out: {why}.]")}]}) + ); + assert_eq!( + read( + declined( + Family::Responses, + &prompt, + Some("cyber"), + false, + prompt.len() + ) + .unwrap() + ) + .1["content"][0]["text"], + "[Another message here is left out, for the same reason.]" + ); + // Shorter wording where the full one does not fit, and none at all + // where nothing does. + let small = Family::Anthropic + .user_item("tell me how to pick this lock") + .unwrap(); + assert_eq!( + read(declined(Family::Anthropic, &small, Some("cyber"), true, small.len()).unwrap()).1 + ["content"][0]["text"], + "[Left out.]" + ); + let tiny = Family::Anthropic.user_item("hi").unwrap(); + assert!( + declined(Family::Anthropic, &tiny, None, true, tiny.len()) + .unwrap() + .is_none() + ); + } + #[test] fn a_result_with_a_picture_reads_as_its_text_and_elides_to_it() { let images = [crate::codec::Image { diff --git a/src/store/db.rs b/src/store/db.rs index bf8c01a0a..b6f1a5735 100644 --- a/src/store/db.rs +++ b/src/store/db.rs @@ -1571,7 +1571,7 @@ impl Database { /// Stored schema version, kept in `PRAGMA user_version`. Stores created /// before versioning and stores from newer binaries are rejected; an older /// versioned store is migrated forward, one version at a time, at open. - pub const SCHEMA: i32 = 60; + pub const SCHEMA: i32 = 61; /// Verbatim user prompts a compaction keeps: per-prompt text, and the /// total text plus `(ordinal, String)` entry metadata. Empty entries cost /// space too, so the retained list cannot grow with conversation length. @@ -1674,6 +1674,8 @@ impl Database { CREATE INDEX IF NOT EXISTS compactions_pinned ON compactions(pinned) WHERE pinned IS NOT NULL; CREATE TABLE IF NOT EXISTS stubs(node INTEGER PRIMARY KEY REFERENCES nodes(id), item BLOB NOT NULL); + CREATE TABLE IF NOT EXISTS declined(node INTEGER PRIMARY KEY REFERENCES nodes(id), + item BLOB NOT NULL); CREATE TABLE IF NOT EXISTS senders(node INTEGER PRIMARY KEY REFERENCES nodes(id), bot TEXT, bot_id INTEGER, turn INTEGER, origin TEXT); CREATE TABLE IF NOT EXISTS elisions(node INTEGER PRIMARY KEY REFERENCES nodes(id), @@ -4436,10 +4438,10 @@ impl Database { // Already single-line items need no rewrite (escaped newlines are fine). // Pictures go without their bytes, as other history reads return them. let mut length = self.conn.prepare_cached(&format!( - "SELECT length({READING_ITEM}) FROM (SELECT {PICTURELESS} AS item FROM nodes WHERE id=?)" + "SELECT length({READING_ITEM}) FROM (SELECT {PICTURELESS} AS item FROM (SELECT COALESCE(d.item,n.item) AS item FROM nodes n LEFT JOIN declined d ON d.node=n.id WHERE n.id=?))" ))?; let mut slice = self.conn.prepare_cached(&format!( - "SELECT substr({READING_ITEM},?,?) FROM (SELECT {PICTURELESS} AS item FROM nodes WHERE id=?)" + "SELECT substr({READING_ITEM},?,?) FROM (SELECT {PICTURELESS} AS item FROM (SELECT COALESCE(d.item,n.item) AS item FROM nodes n LEFT JOIN declined d ON d.node=n.id WHERE n.id=?))" ))?; let mut text = String::new(); let mut position = 0_u64; @@ -4539,13 +4541,15 @@ impl Database { // view has may not be in the item as stored: the view decides. let mut contains = self.conn.prepare_cached( "SELECT instr(item,x'0a')>0 OR instr(item,x'0d')>0 - OR instr(lower(CAST(item AS TEXT)),?2)>0 FROM nodes WHERE id=?1", + OR instr(lower(CAST(item AS TEXT)),?2)>0 + FROM (SELECT COALESCE(d.item,n.item) AS item FROM nodes n LEFT JOIN declined d ON d.node=n.id WHERE n.id=?1)", )?; // The raw item may match only in what the view leaves out, such as // a picture's bytes, so a match counts once the view has it too. let mut reading = self.conn.prepare_cached(&format!( "SELECT CAST({READING_ITEM} AS TEXT),json_extract(CAST(item AS TEXT),'$.type') - FROM (SELECT {PICTURELESS} AS item FROM nodes WHERE id=?)" + FROM (SELECT {PICTURELESS} AS item FROM + (SELECT COALESCE(d.item,n.item) AS item FROM nodes n LEFT JOIN declined d ON d.node=n.id WHERE n.id=?))" ))?; let mut rows = chain.query(params![head])?; let mut skipping = own || before.is_some(); @@ -4710,7 +4714,8 @@ impl Database { if let Some((turn, mut saved, status, saved_options, prompt_node, saved_images)) = prior { if let Some(node) = prompt_node { saved = self.conn.prepare_cached( - "SELECT json_extract(CAST(item AS TEXT),'$.content[0].text') FROM nodes WHERE id=?" + "SELECT json_extract(CAST(COALESCE((SELECT item FROM declined WHERE node=?1),item) AS TEXT), + '$.content[0].text') FROM nodes WHERE id=?1" )?.query_row([node], |r| r.get(0))?; } // A turn records the folder it runs in; a retry that names none @@ -7033,6 +7038,72 @@ impl Database { "UPDATE turns SET status=?,finished_ms=? WHERE id=?", params![status, epoch_ms(), turn], )?; + // A request a safety classifier declined, and the bot's fallback + // too: what it added since the model last answered goes to later + // requests as notes in its place, since Anthropic says to remove it + // before going on. Each note is the size of what it replaces, so + // nothing counted for the context changes, and the originals are + // kept beside them. + let mut left_out = 0; + if code == Some("provider_refusal") { + let category = (error.and_then(|e| e.facts.as_ref())) + .and_then(|facts| facts.get("category")) + .and_then(Value::as_str); + let family = bot.family()?; + let mut added = Vec::new(); + let (mut node, mut child) = (head, None::); + while let Some(id) = node { + let (parent, item, first): (Option, Vec, Option) = tx + .prepare_cached("SELECT parent,item,turn FROM nodes WHERE id=?")? + .query_row([id], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?; + if super::context::model_output(&item) { + break; + } + // A node another bot reaches, as a fork taken while the + // request was out does, is its history too: it and what + // came before it stay as they are. + let shared: bool = tx + .prepare_cached( + "SELECT EXISTS(SELECT 1 FROM bots WHERE name!=?2 AND (head=?1 OR context_start=?1 + OR note=?1 OR compaction=?1 OR elision=?1)) + OR EXISTS(SELECT 1 FROM nodes WHERE parent=?1 AND id IS NOT ?3)", + )? + .query_row(params![id, bot.name, child], |r| r.get(0))?; + if shared { + break; + } + added.push((id, item)); + if first == Some(turn) { + break; + } + (node, child) = (parent, Some(id)); + } + let mut noted = false; + for (id, item) in added.iter().rev() { + let result = super::context::is_tool_result(item); + let first = !result && !noted; + let note = super::context::declined(family, item, category, first, item.len())?; + let Some(note) = note else { continue }; + noted |= !result; + left_out += 1; + tx.prepare_cached("INSERT OR IGNORE INTO declined(node,item) VALUES (?,?)")? + .execute(params![id, item])?; + tx.prepare_cached("UPDATE nodes SET item=? WHERE id=?")? + .execute(params![note, id])?; + // A stub quotes the result, so it says the same. + let stub: Option = tx + .prepare_cached("SELECT length(item) FROM stubs WHERE node=?")? + .query_row([id], |r| r.get(0)) + .optional()?; + if let Some(len) = stub + && let Some(note) = + super::context::declined(family, item, category, first, len as usize)? + { + tx.prepare_cached("UPDATE stubs SET item=? WHERE node=?")? + .execute(params![note, id])?; + } + } + } // An expiring bot's clock starts again from this turn's end. tx.execute( "UPDATE bots SET head=?,running_turn=NULL,status=?,closed=NULL,open_calls=NULL, @@ -7049,6 +7120,9 @@ impl Database { let mut data = json!({"status":status,"checkpoint":if status == "completed" { head } else { None }, "error":code,"detail":error.and_then(|e| e.detail.clone())}); with_facts(&mut data, error); + if left_out > 0 { + data["left_out"] = json!(left_out); + } let cursor = event(&tx, &bot.name, Some(turn), "turn_finished", data.clone())?; tx.commit()?; self.approvals -= unstarted; @@ -8109,6 +8183,7 @@ impl Database { tx.execute("DELETE FROM compactions WHERE node=?", [id])?; tx.execute("DELETE FROM elisions WHERE node=?", [id])?; tx.execute("DELETE FROM stubs WHERE node=?", [id])?; + tx.execute("DELETE FROM declined WHERE node=?", [id])?; tx.execute("DELETE FROM senders WHERE node=?", [id])?; tx.execute("DELETE FROM nodes WHERE id=?", [id])?; freed += 1; @@ -8432,7 +8507,8 @@ impl Database { // bound is on the item as read, its pictures left out. let mut query = self.conn.prepare_cached(&format!( "SELECT CASE WHEN length(n.item)<=?2 THEN n.item END,s.bot,s.turn,s.bot_id,s.origin - FROM (SELECT id,{PICTURELESS} AS item FROM nodes WHERE id=?1) n + FROM (SELECT id,{PICTURELESS} AS item FROM + (SELECT n.id,COALESCE(d.item,n.item) AS item FROM nodes n LEFT JOIN declined d ON d.node=n.id WHERE n.id=?1)) n LEFT JOIN senders s ON s.node=n.id" ))?; for &node in wanted { @@ -8530,12 +8606,13 @@ impl Database { let mut statement = self.conn.prepare_cached(&format!( "SELECT {TURN_VIEW},COALESCE(t.workspace,b.workspace), COALESCE(t.model,t.bot_model,b.provider||'/'||b.model), - substr(CASE WHEN t.prompt_node IS NULL THEN t.prompt ELSE json_extract(CAST(n.item AS TEXT),'$.content[0].text') END,1,200), - length(CASE WHEN t.prompt_node IS NULL THEN t.prompt ELSE json_extract(CAST(n.item AS TEXT),'$.content[0].text') END), + substr(CASE WHEN t.prompt_node IS NULL THEN t.prompt ELSE json_extract(CAST(COALESCE(d.item,n.item) AS TEXT),'$.content[0].text') END,1,200), + length(CASE WHEN t.prompt_node IS NULL THEN t.prompt ELSE json_extract(CAST(COALESCE(d.item,n.item) AS TEXT),'$.content[0].text') END), t.delivery,COALESCE(t.reasoning,t.inherited_reasoning, CASE WHEN t.bot_model IS NULL THEN b.reasoning ELSE t.bot_reasoning END),t.summary_ms, t.budget_tokens FROM turns t JOIN bots b ON b.name=t.bot LEFT JOIN nodes n ON n.id=t.prompt_node + LEFT JOIN declined d ON d.node=t.prompt_node WHERE t.bot=?1 AND t.id>?2 AND t.id = self.conn - .query_row("SELECT item FROM nodes WHERE id=?", [node], |r| r.get(0))?; + .query_row( + "SELECT COALESCE(d.item,n.item) FROM nodes n LEFT JOIN declined d ON d.node=n.id WHERE n.id=?", + [node], + |r| r.get(0), + )?; let Some((_, _, output)) = super::context::tool_result(&item) else { return fail("result_not_found"); }; @@ -10346,6 +10427,12 @@ fn migrate(conn: &Connection, from: i32) -> Result<()> { ALTER TABLE bots DROP COLUMN fallbacks;", )?; } + // 60 -> 61: what a declined request added goes to later requests as + // notes. No request was declined so. + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS declined(node INTEGER PRIMARY KEY REFERENCES nodes(id), + item BLOB NOT NULL);", + )?; Ok(()) } /// 53 -> 54: a turn's denial count became a window of its last decided diff --git a/tests/test_runtime.py b/tests/test_runtime.py index d3bb274e9..a1611f229 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -780,9 +780,13 @@ def do_POST(self): # as a reply after a reaction with nothing to add. if last['content'][0]['content'].startswith('silent'): blocks = [] + elif last['content'][0]['content'].startswith('refuse:all'): + # What a tool read sets the classifier off. + stop = 'refusal' + details = {'type': 'refusal', 'category': 'cyber', 'explanation': None} else: blocks.append({'type': 'text', 'text': 'echo:' + last['content'][0]['content']}) - stop = 'end_turn' + stop = 'refusal' if details else 'end_turn' else: user = last['content'][0]['text'] if user == 'think-only': @@ -1193,6 +1197,65 @@ def test_a_refusal_the_fallback_repeats_fails_the_turn(self): self.assertEqual(client.request('create', bot='Bad', workspace=str(path), fallback='openai/x')['error'], 'provider_unavailable') + def test_what_a_declined_request_added_goes_on_as_a_note(self): + client, model, path = self.start() + client.request('create', bot='Bob', workspace=str(path), effort='low') + why = "the provider's safety classifier declined the request that carried it (category: cyber)" + # Declined at its first request: the prompt is what it added. + words = 'refuse:all ' + 'word ' * 40 + turn = client.request('submit', bot='Bob', request_id='d1', prompt=words)['result']['turn'] + finished = client.finished(turn)['data'] + self.assertEqual(finished['error'], 'provider_refusal') + turn = client.request('submit', bot='Bob', request_id='d2', prompt='hello')['result']['turn'] + self.assertEqual(client.finished(turn)['data']['status'], 'completed') + model.requests.get(timeout=1) + after = model.requests.get(timeout=1) + self.assertEqual(after['messages'][0]['content'], + [{'type': 'text', 'text': f'[A message here is left out: {why}.]'}]) + self.assertEqual(after['messages'][1]['content'], [{'type': 'text', 'text': 'hello'}]) + # Declined after a tool ran: only its result goes as a note; the + # prompt and the call the model made stay. + read = 'tool:refuse:all ' + 'line ' * 40 + turn = client.request('submit', bot='Bob', request_id='d3', prompt=read)['result']['turn'] + self.assertEqual(client.finished(turn)['data']['error'], 'provider_refusal') + turn = client.request('submit', bot='Bob', request_id='d4', prompt='again')['result']['turn'] + self.assertEqual(client.finished(turn)['data']['status'], 'completed') + while not model.requests.empty(): + last = model.requests.get(timeout=1) + sent = [m['content'] for m in last['messages']] + self.assertEqual(sent[-4][0]['text'], read) + self.assertEqual(sent[-3][1]['input'], {'text': read[5:]}) + self.assertEqual(sent[-2], [{'type': 'tool_result', 'tool_use_id': 'toolu_1', + 'content': f'[This tool result is left out: {why}.]'}]) + self.assertEqual(sent[-1], [{'type': 'text', 'text': 'again'}]) + # The history reads give the originals back; only the model gets notes. + with closing(sqlite3.connect(path / 'state.sqlite')) as db: + kept = [row[0] for row in db.execute('SELECT node FROM declined ORDER BY node')] + items = [node_item(client, 'Bob', node)['result'] for node in kept] + self.assertEqual(items[0]['content'][0]['text'], words) + self.assertEqual(items[1]['content'][0]['content'], read[5:]) + # The person's words stay in the record, and a retry still matches them. + turns = client.request('turns', bot='Bob', after=0, limit=8)['result']['turns'] + self.assertEqual(turns[0]['prompt_preview'], words[:200]) + again = client.request('submit', bot='Bob', request_id='d1', prompt=words) + self.assertEqual(again['result']['turn'], turns[0]['turn'], again) + + def test_a_fork_taken_while_a_request_is_out_keeps_what_it_declined(self): + client, model, path = self.start() + client.request('create', bot='Bob', workspace=str(path), effort='low') + # The fork takes the prompt while the request is out; the refusal + # that follows is Bob's, not the fork's. + model.generate_delay = 1 + words = 'refuse:all ' + 'word ' * 40 + turn = client.request('submit', bot='Bob', request_id='d1', prompt=words)['result']['turn'] + model.requests.get(timeout=5) + self.assertIn('result', client.request('fork', source='Bob', bot='Branch')) + self.assertEqual(client.finished(turn)['data']['error'], 'provider_refusal') + turn = client.request('submit', bot='Branch', request_id='b1', prompt='hello')['result']['turn'] + self.assertEqual(client.finished(turn)['data']['status'], 'completed') + sent = model.requests.get(timeout=1) + self.assertEqual(sent['messages'][0]['content'], [{'type': 'text', 'text': words}]) + def test_a_hosted_search_is_declared_kept_and_its_paused_turn_resumed(self): client, model, path = self.start() self.assertTrue(client.ready['providers']['anthropic']['web_search'])