Skip to content

feat(cli, config, conversation): Add --cfg reset keywords #849

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
JeanMertz wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(cli, config, conversation): Add --cfg reset keywords #849

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
80 changes: 71 additions & 9 deletions crates/jp_cli/src/cmd/query.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ use std::{
};

use camino::{Utf8Path, Utf8PathBuf};
use chrono::{DateTime, Utc};
use clap::{ArgAction, builder::TypedValueParser as _};
use indexmap::IndexMap;
use jp_attachment::Attachment;
Expand All @@ -85,6 +86,7 @@ use jp_config::{
use jp_conversation::{
Conversation, ConversationEvent, ConversationId, ConversationStream,
event::{ChatRequest, ChatResponse},
stream::{ApplyDelta, ResetDelta},
thread::{Thread, ThreadBuilder},
};
use jp_inquire::prompt::TerminalPromptBackend;
Expand Down Expand Up @@ -123,6 +125,7 @@ use crate::{
conversation::fork,
lock::{LockRequest, acquire_lock},
},
config_pipeline::{ConfigReset, ConfigResetEvents},
ctx::IntoPartialAppConfig,
editor,
error::{Error, Result},
Expand Down Expand Up @@ -361,7 +364,7 @@ impl Query {
// 2. picker "start new": `start_new` is set, create a fresh conversation.
// 3. --fork/--id/session: resolve an existing conversation, lock it.
// 4. Lock contention: user picks "new" or "fork" from the prompt.
let lock = self.acquire_lock(ctx, handle, start_new).await?;
let (lock, fresh) = self.acquire_lock(ctx, handle, start_new).await?;

// Create symlinks and seed approvals for any `--mount` flags before the
// turn runs, so tools can reach the mounted paths.
Expand All @@ -381,7 +384,29 @@ impl Query {
warn!(%error, "Failed to record activation.");
}

if let Some(delta) = get_config_delta_from_cli(&cfg, &lock)? {
// Persist config state changes into the conversation stream.
//
// A fresh conversation needs neither branch: its base config was
// written from this invocation's resolved config at creation time
// (that also absorbs any `--cfg` reset keyword, per [RFD 038]).
//
// A conversation carrying earlier config state — continuing or forked
// — records a `--cfg` reset keyword as its stream events, appended
// directly: between the `Reset` and whichever `Apply` restores the
// required fields the stream does not resolve to a valid config, so
// the empty-diff suppression path in `add_config_delta` cannot run.
//
// Without a reset keyword, any divergence between the stream's config
// and this invocation's resolved config is appended as a single
// suppression-checked `Apply` diff, as before.
//
// [RFD 038]: https://jp.computer/rfd/038
if let Some(reset_events) = ctx.config_reset.take() {
if !fresh {
lock.as_mut()
.update_events(|events| persist_config_reset(events, reset_events, now));
}
} else if let Some(delta) = get_config_delta_from_cli(&cfg, &lock)? {
lock.as_mut()
.update_events(|events| events.add_config_delta(delta));
}
Expand Down Expand Up @@ -964,15 +989,22 @@ impl Query {
Ok(())
}

/// Resolve the target conversation and return its exclusive lock.
///
/// The second element is `true` when the conversation was freshly created
/// by this call: its base config is this invocation's resolved config, so
/// no config state predates it.
/// Forks return `false` — a fork copies the source's base config and
/// events, and therefore carries config state from before this invocation.
async fn acquire_lock(
&self,
ctx: &mut Ctx,
handle: Option<ConversationHandle>,
start_new: bool,
) -> Result<ConversationLock> {
) -> Result<(ConversationLock, bool)> {
// Handle --new: create a fresh conversation.
if self.is_new() {
return self.create_new_conversation(ctx);
return Ok((self.create_new_conversation(ctx)?, true));
}

// Handle the picker's "start a new conversation" choice. It carries no
Expand All @@ -982,24 +1014,26 @@ impl Query {
if !self.allows_new_from_picker() {
return Err(Error::NewConflictsWithTarget);
}
return self.create_new_conversation(ctx);
return Ok((self.create_new_conversation(ctx)?, true));
}

let handle = handle.ok_or(Error::NoConversationTarget)?;

// Handle --fork: fork the conversation before locking.
if let Some(fork_turns) = &self.fork {
return fork_conversation(ctx, &handle, *fork_turns);
return Ok((fork_conversation(ctx, &handle, *fork_turns)?, false));
}

let req = LockRequest::from_ctx(handle, ctx)
.allow_new(true)
.allow_fork(true);

match acquire_lock(req).await? {
LockOutcome::Acquired(lock) => Ok(lock),
LockOutcome::NewConversation => self.create_new_conversation(ctx),
LockOutcome::ForkConversation(handle) => fork_conversation(ctx, &handle, None),
LockOutcome::Acquired(lock) => Ok((lock, false)),
LockOutcome::NewConversation => Ok((self.create_new_conversation(ctx)?, true)),
LockOutcome::ForkConversation(handle) => {
Ok((fork_conversation(ctx, &handle, None)?, false))
}
}
}
}
Expand Down Expand Up @@ -1236,6 +1270,34 @@ fn apply_title_override(lock: &ConversationLock, title: Option<&str>, no_title:
}
}

/// Append a `--cfg` reset keyword's events to a conversation stream.
///
/// Persists the reset-then-layer sequence from [RFD 038]: a [`ResetDelta`]
/// marking the reset point, then the workspace partial for `WORKSPACE` resets,
/// then whatever state this invocation layered on top of the reset point.
/// Empty layers are skipped by [`ConversationStream::add_config_reset`], which
/// also documents why the sequence bypasses diff-suppression.
///
/// [RFD 038]: https://jp.computer/rfd/038
fn persist_config_reset(
events: &mut ConversationStream,
reset: ConfigResetEvents,
timestamp: DateTime<Utc>,
) {
let mut layers = Vec::with_capacity(2);

if let ConfigReset::Workspace(delta) = reset.reset {
layers.push(ApplyDelta { timestamp, delta });
}

layers.push(ApplyDelta {
timestamp,
delta: reset.post,
});

events.add_config_reset(ResetDelta { timestamp }, layers);
}

fn get_config_delta_from_cli(
cfg: &AppConfig,
lock: &ConversationLock,
Expand Down
137 changes: 136 additions & 1 deletion crates/jp_cli/src/cmd/query_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ fn build_query_config(
query: &Query,
handle: Option<&ConversationHandle>,
) -> AppConfig {
let pipeline = ConfigPipeline::new(base, cfg_args, Some(workspace), None).unwrap();
let pipeline = ConfigPipeline::new(cfg_args, Some(workspace), None, || Ok(base)).unwrap();

let conversation_partial = handle.map(|handle| {
query
Expand Down Expand Up @@ -754,6 +754,141 @@ fn query_cfg_sourced_compaction_persists_as_config_delta() {
);
}

/// The `config_delta` events of a stream's serialized `events.json`.
fn serialized_config_deltas(events: &ConversationStream) -> Vec<Value> {
let (_base, serialized) = events.clone().to_parts().unwrap();
serialized
.into_iter()
.filter(|event| event.get("type").and_then(Value::as_str) == Some("config_delta"))
.collect()
}

#[test]
fn cfg_reset_none_appends_reset_then_post_apply() {
// `jp q --cfg=NONE --cfg=<post>` on a continuing conversation ([RFD 038]):
// the stream records `[Reset, Apply(post)]`, where `post` restores the
// required fields on top of program defaults.
let base_config = Arc::new(config_with_model(ProviderId::Anthropic, "base-model"));
let conversation_id = make_id(3000);

let mut workspace = Workspace::new("/tmp/test");
workspace.create_conversation_with_id(
conversation_id,
Conversation::default(),
Arc::clone(&base_config),
);
let handle = workspace.acquire_conversation(&conversation_id).unwrap();
let lock = workspace.test_lock(handle);

let post = config_with_model(ProviderId::Openai, "fresh-model").to_partial();
let reset = ConfigResetEvents {
reset: ConfigReset::Defaults,
post: Box::new(post),
};

lock.as_mut()
.update_events(|events| persist_config_reset(events, reset, DateTime::<Utc>::UNIX_EPOCH));

let events = lock.events().clone();

// The stream resolves to the post-reset state, not the pre-reset base.
let merged = events.config().unwrap();
let model_id = merged.assistant.model.id.resolved();
assert_eq!(model_id.provider, ProviderId::Openai);
assert_eq!(model_id.name.as_ref(), "fresh-model");

// Wire shape: a `Reset` marker followed by a plain `Apply` (no `op`).
let deltas = serialized_config_deltas(&events);
assert_eq!(deltas.len(), 2, "expected [Reset, Apply], got {deltas:?}");
assert_eq!(deltas[0].get("op").and_then(Value::as_str), Some("reset"));
assert!(
deltas[1].get("op").is_none(),
"`Apply` writes no `op` field"
);
}

#[test]
fn cfg_reset_workspace_appends_reset_then_workspace_apply() {
// `jp q --cfg=WORKSPACE` on a continuing conversation ([RFD 038]): the
// stream records `[Reset, Apply(workspace)]`, re-adopting the workspace's
// resolved config as of this invocation. No further `Apply` is written
// when nothing is layered on top.
let base_config = Arc::new(config_with_model(ProviderId::Anthropic, "base-model"));
let conversation_id = make_id(3001);

let mut workspace = Workspace::new("/tmp/test");
workspace.create_conversation_with_id(
conversation_id,
Conversation::default(),
Arc::clone(&base_config),
);
let handle = workspace.acquire_conversation(&conversation_id).unwrap();
let lock = workspace.test_lock(handle);

let workspace_partial = config_with_model(ProviderId::Openai, "ws-model").to_partial();
let reset = ConfigResetEvents {
reset: ConfigReset::Workspace(Box::new(workspace_partial)),
post: Box::new(PartialAppConfig::default()),
};

lock.as_mut()
.update_events(|events| persist_config_reset(events, reset, DateTime::<Utc>::UNIX_EPOCH));

let events = lock.events().clone();

let merged = events.config().unwrap();
let model_id = merged.assistant.model.id.resolved();
assert_eq!(model_id.provider, ProviderId::Openai);
assert_eq!(model_id.name.as_ref(), "ws-model");

let deltas = serialized_config_deltas(&events);
assert_eq!(
deltas.len(),
2,
"empty post-reset state must not write a third event: {deltas:?}"
);
assert_eq!(deltas[0].get("op").and_then(Value::as_str), Some("reset"));
assert!(deltas[1].get("op").is_none());
}

#[test]
fn cfg_reset_none_without_post_leaves_unresolvable_config() {
// A bare `jp q --cfg=NONE` on a continuing conversation records only the
// `Reset`. Program defaults lack required fields (`assistant.model.id`),
// so resolving the stream's config fails until a later `Apply` restores
// them — the intended escape-hatch semantics from [RFD 038].
let base_config = Arc::new(config_with_model(ProviderId::Anthropic, "base-model"));
let conversation_id = make_id(3002);

let mut workspace = Workspace::new("/tmp/test");
workspace.create_conversation_with_id(
conversation_id,
Conversation::default(),
Arc::clone(&base_config),
);
let handle = workspace.acquire_conversation(&conversation_id).unwrap();
let lock = workspace.test_lock(handle);

let reset = ConfigResetEvents {
reset: ConfigReset::Defaults,
post: Box::new(PartialAppConfig::default()),
};

lock.as_mut()
.update_events(|events| persist_config_reset(events, reset, DateTime::<Utc>::UNIX_EPOCH));

let events = lock.events().clone();

let deltas = serialized_config_deltas(&events);
assert_eq!(deltas.len(), 1, "expected only the Reset: {deltas:?}");
assert_eq!(deltas[0].get("op").and_then(Value::as_str), Some("reset"));

assert!(
events.config().is_err(),
"program defaults alone must fail validation"
);
}

#[test]
fn inline_compact_dsl_is_not_written_into_query_config() {
// The inline `-k SPEC` plan is applied as overlay events at query time, not
Expand Down
Loading
Loading