mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-21 08:02:38 +00:00
Merge remote-tracking branch 'origin/main' into hugo/agent-memory-refactor
# Conflicts: # backend/ee-repo-ref.txt # frontend/src/lib/components/InstanceSettings.svelte
This commit is contained in:
hugocasa
@@ -1 +1 @@
|
||||
12551a748d759c7c45a9da39b63117a059eba845
|
||||
1a8f29cd14fce3b7c6eb20d27f5b58b8ae56d9f5
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
ALTER TABLE workspace_settings DROP COLUMN IF EXISTS ai_sessions_backup_generation;
|
||||
@@ -0,0 +1,4 @@
|
||||
-- Bumped by every workspace key rotation: the AI session backups in the workspace storage
|
||||
-- live under a prefix named by it, so a rotation moves to a fresh prefix and the previous
|
||||
-- ones can be deleted at leisure without ever touching live objects.
|
||||
ALTER TABLE workspace_settings ADD COLUMN ai_sessions_backup_generation BIGINT NOT NULL DEFAULT 0;
|
||||
@@ -234,7 +234,7 @@ workspace_protection_rule: workspace_id(char), name(char), rules(int), bypass_gr
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
workspace_runnable_dependencies: flow_path(char), runnable_path(char), script_hash(bigint), runnable_is_flow(bool), workspace_id(char), app_path(char), id(bigint)
|
||||
FK: (app_path, workspace_id) -> app(path, workspace_id) | (flow_path, workspace_id) -> flow(path, workspace_id)
|
||||
workspace_settings: workspace_id(char), slack_team_id(char), slack_name(char), slack_command_script(char), slack_email(char), customer_id(char), plan(char), webhook(text), ai_config(jsonb), large_file_storage(jsonb), git_sync(jsonb), default_app(char), default_scripts(jsonb), deploy_ui(jsonb), mute_critical_alerts(bool), color(char), operator_settings(jsonb), teams_command_script(text), teams_team_id(text), teams_team_name(text), git_app_installations(jsonb), ducklake(jsonb), slack_oauth_client_id(char), slack_oauth_client_secret(char), datatable(jsonb), teams_team_guid(text), auto_invite(jsonb), error_handler(jsonb), success_handler(jsonb), public_app_execution_limit_per_minute(int), dbt_warehouses(jsonb), guest_access_enabled(bool), guest_jwt_public_key(text), guest_jwt_jwks_url(text)
|
||||
workspace_settings: workspace_id(char), slack_team_id(char), slack_name(char), slack_command_script(char), slack_email(char), customer_id(char), plan(char), webhook(text), ai_config(jsonb), large_file_storage(jsonb), git_sync(jsonb), default_app(char), default_scripts(jsonb), deploy_ui(jsonb), mute_critical_alerts(bool), color(char), operator_settings(jsonb), teams_command_script(text), teams_team_id(text), teams_team_name(text), git_app_installations(jsonb), ducklake(jsonb), slack_oauth_client_id(char), slack_oauth_client_secret(char), datatable(jsonb), teams_team_guid(text), auto_invite(jsonb), error_handler(jsonb), success_handler(jsonb), public_app_execution_limit_per_minute(int), dbt_warehouses(jsonb), guest_access_enabled(bool), guest_jwt_public_key(text), guest_jwt_jwks_url(text), ai_sessions_backup_generation(int)
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
zombie_job_counter: job_id(uuid), counter(int)
|
||||
FK: (job_id) -> v2_job(id)
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,150 @@
|
||||
//! What the workspace key rotation and the AI session backup routes
|
||||
//! (`windmill-api/src/ai_sessions.rs`) share about the backups in the workspace storage.
|
||||
//!
|
||||
//! The backups are ciphertext under the workspace key and live under a prefix named by a
|
||||
//! generation the rotation bumps (`workspace_settings.ai_sessions_backup_generation`) in the
|
||||
//! transaction that commits the new key. A rotation does not re-key them: once committed,
|
||||
//! the routes read and write under the new generation's prefix and answer with its number
|
||||
//! (`backup_generation`; the storage identity, `storage_id`, names the storage and does not
|
||||
//! change), so every browser marks its sync state stale and pushes its sessions whole again
|
||||
//! there, and every older generation, which nothing writes to any
|
||||
//! more, is deleted off the request at leisure. Sessions no browser holds any more are lost,
|
||||
//! which a rotation (a rare operation) accepts in exchange for having no key but the current
|
||||
//! one to read with and nothing to rewrite in place. A generation is never reused, so no
|
||||
//! deletion, however late, can touch live objects; a rotation that fails before its commit
|
||||
//! bumps nothing and deletes nothing; two rotations racing serialize on the key row.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use futures::TryStreamExt;
|
||||
use windmill_common::error::{Error, Result};
|
||||
use windmill_common::utils::calculate_hash;
|
||||
use windmill_common::DB;
|
||||
use windmill_object_store::object_store_reexports::{
|
||||
ObjectStore, ObjectStoreError, Path as ObjectPath,
|
||||
};
|
||||
use windmill_object_store::{object_store_error_to_error, ObjectStoreResource};
|
||||
use windmill_types::s3::LargeFileStorage;
|
||||
|
||||
/// The root of every AI session backup key in a workspace's storage.
|
||||
pub const ROOT: &str = "windmill_ai_sessions";
|
||||
/// The push body cap: no object written through the routes is larger. One that is was
|
||||
/// planted by whoever holds the bucket's credentials, and is left unread.
|
||||
pub const MAX_OBJECT_BYTES: usize = 32 * 1024 * 1024;
|
||||
|
||||
const IO_CONCURRENCY: usize = 8;
|
||||
|
||||
/// The prefix of one generation's objects: `windmill_ai_sessions/{w_id}/g{generation}/`.
|
||||
pub fn generation_prefix(w_id: &str, generation: i64) -> String {
|
||||
format!("{ROOT}/{w_id}/g{generation}")
|
||||
}
|
||||
|
||||
/// Names the storage the backups are in, by what locates its objects (endpoint, region,
|
||||
/// bucket; never the credentials, which rotate), so a browser tells that its sync state was
|
||||
/// recorded against another storage; the generation, answered alongside, tells it a
|
||||
/// rotation happened in this one.
|
||||
pub fn storage_id(resource: &ObjectStoreResource) -> String {
|
||||
let location = match resource {
|
||||
ObjectStoreResource::S3(s) => format!(
|
||||
"s3:{}:{}:{}:{}",
|
||||
s.endpoint,
|
||||
s.port.unwrap_or_default(),
|
||||
s.region,
|
||||
s.bucket
|
||||
),
|
||||
ObjectStoreResource::Azure(a) => format!(
|
||||
"azure:{}:{}:{}",
|
||||
a.endpoint.as_deref().unwrap_or_default(),
|
||||
a.account_name,
|
||||
a.container_name
|
||||
),
|
||||
ObjectStoreResource::Gcs(g) => format!("gcs:{}", g.bucket),
|
||||
ObjectStoreResource::Filesystem(f) => format!("fs:{}", f.root_path),
|
||||
};
|
||||
calculate_hash(&location)[..16].to_string()
|
||||
}
|
||||
|
||||
/// The workspace's primary storage, resolved without a caller: a rotation runs the
|
||||
/// deletion off its own request.
|
||||
async fn primary_store(db: &DB, w_id: &str) -> Result<Option<Arc<dyn ObjectStore>>> {
|
||||
let Some(lfs_json) = sqlx::query_scalar!(
|
||||
"SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1",
|
||||
w_id
|
||||
)
|
||||
.fetch_optional(db)
|
||||
.await?
|
||||
.flatten() else {
|
||||
return Ok(None);
|
||||
};
|
||||
let lfs: LargeFileStorage = serde_json::from_value(lfs_json)
|
||||
.map_err(|e| Error::internal_err(format!("parsing large_file_storage: {e}")))?;
|
||||
let resource_value = if matches!(lfs, LargeFileStorage::FilesystemStorage(_)) {
|
||||
serde_json::Value::Null
|
||||
} else {
|
||||
let path = lfs.get_s3_resource_path();
|
||||
let path = path.strip_prefix("$res:").unwrap_or(path);
|
||||
windmill_common::workspaces::transform_json_value_unchecked(
|
||||
&serde_json::Value::String(format!("$res:{path}")),
|
||||
w_id,
|
||||
db,
|
||||
)
|
||||
.await?
|
||||
};
|
||||
let resource = windmill_object_store::lfs_to_object_store_resource(&lfs, resource_value)?;
|
||||
Ok(Some(
|
||||
windmill_object_store::build_object_store_client(&resource).await?,
|
||||
))
|
||||
}
|
||||
|
||||
/// The generation an object key sits under, `None` for a key of no generation (an older
|
||||
/// layout), which counts as older than any.
|
||||
fn generation_of(w_id: &str, key: &ObjectPath) -> Option<i64> {
|
||||
key.as_ref()
|
||||
.strip_prefix(&format!("{ROOT}/{w_id}/g"))?
|
||||
.split('/')
|
||||
.next()?
|
||||
.parse()
|
||||
.ok()
|
||||
}
|
||||
|
||||
/// Deletes, off the request and as the listing streams, every object of the workspace's
|
||||
/// backups from a generation older than `current`, once the rotation that made `current`
|
||||
/// the generation has committed: nothing writes there any more but a push that resolved its
|
||||
/// prefix before the commit, junk the browser's next push of that session rewrites under the
|
||||
/// current prefix, as is anything a deletion cut short left behind. For the rotation route,
|
||||
/// which authorized its caller as a superadmin.
|
||||
pub(crate) fn spawn_delete_older(db: DB, w_id: String, current: i64) {
|
||||
tokio::spawn(async move {
|
||||
let store = match primary_store(&db, &w_id).await {
|
||||
Ok(Some(store)) => store,
|
||||
Ok(None) => return,
|
||||
Err(e) => {
|
||||
tracing::warn!("older AI session backups of {w_id} left in place: {e:#}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let prefix = ObjectPath::from(format!("{ROOT}/{w_id}"));
|
||||
let deleted = store
|
||||
.list(Some(&prefix))
|
||||
.map_err(object_store_error_to_error)
|
||||
.try_for_each_concurrent(IO_CONCURRENCY, |meta| {
|
||||
let (store, w_id) = (&store, &w_id);
|
||||
async move {
|
||||
if generation_of(w_id, &meta.location).is_some_and(|g| g >= current) {
|
||||
return Ok(());
|
||||
}
|
||||
match store.delete(&meta.location).await {
|
||||
Ok(()) | Err(ObjectStoreError::NotFound { .. }) => Ok(()),
|
||||
Err(e) => Err(object_store_error_to_error(e)),
|
||||
}
|
||||
}
|
||||
})
|
||||
.await;
|
||||
match deleted {
|
||||
Ok(()) => {
|
||||
tracing::info!("deleted the AI session backups of {w_id} older than g{current}")
|
||||
}
|
||||
Err(e) => tracing::warn!("deleting the older AI session backups of {w_id}: {e:#}"),
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -1,6 +1,8 @@
|
||||
#[cfg(feature = "parquet")]
|
||||
pub mod ai_session_backups;
|
||||
pub mod data_metrics;
|
||||
pub mod datatable_migrations;
|
||||
pub mod deployment_requests;
|
||||
pub mod data_metrics;
|
||||
pub mod workspaces;
|
||||
pub mod workspaces_extra;
|
||||
pub mod workspaces_oss;
|
||||
|
||||
@@ -5339,6 +5339,17 @@ async fn set_encryption_key(
|
||||
|
||||
let mut tx = db.begin().await?;
|
||||
|
||||
// Under the row's lock, so two rotations racing serialize and each sees the key the
|
||||
// other committed. The AI session backups in the workspace storage live under a prefix
|
||||
// named by a generation this bumps (with the key, in this transaction) rather than
|
||||
// being re-keyed; the older generations are deleted once this one has committed (see
|
||||
// `ai_session_backups`). The same key set again is no rotation to them.
|
||||
let previous_key: String = sqlx::query_scalar(
|
||||
"SELECT key FROM workspace_key WHERE workspace_id = $1 AND kind = 'cloud' FOR UPDATE",
|
||||
)
|
||||
.bind(&w_id)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
sqlx::query!(
|
||||
"UPDATE workspace_key SET key = $1 WHERE workspace_id = $2",
|
||||
request.new_key.clone(),
|
||||
@@ -5346,6 +5357,18 @@ async fn set_encryption_key(
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
let backups_generation: Option<i64> = if previous_key != request.new_key {
|
||||
sqlx::query_scalar(
|
||||
"UPDATE workspace_settings SET ai_sessions_backup_generation = \
|
||||
ai_sessions_backup_generation + 1 WHERE workspace_id = $1 \
|
||||
RETURNING ai_sessions_backup_generation",
|
||||
)
|
||||
.bind(&w_id)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let mut reencrypted_secret_paths: Vec<String> = Vec::new();
|
||||
if !request.skip_reencrypt.unwrap_or(false) {
|
||||
@@ -5402,6 +5425,15 @@ async fn set_encryption_key(
|
||||
// Invalidate the cache only after the transaction has committed
|
||||
WORKSPACE_CRYPT_CACHE.remove(w_id.as_str());
|
||||
|
||||
// Nothing writes under the older generations any more; the browsers push their
|
||||
// sessions again under the new one.
|
||||
#[cfg(feature = "parquet")]
|
||||
if let Some(generation) = backups_generation {
|
||||
crate::ai_session_backups::spawn_delete_older(db.clone(), w_id.clone(), generation);
|
||||
}
|
||||
#[cfg(not(feature = "parquet"))]
|
||||
let _ = backups_generation;
|
||||
|
||||
// Build the batch: one event for the encryption key itself plus one per
|
||||
// re-encrypted secret variable. The batch entrypoint dispatches a single
|
||||
// git-sync job per repo carrying all items, so repos with Secrets sync
|
||||
|
||||
@@ -13188,6 +13188,153 @@ paths:
|
||||
type: boolean
|
||||
description: more buckets matched than were returned, so summing them under-reports
|
||||
|
||||
/w/{workspace}/ai/sessions/list:
|
||||
get:
|
||||
summary: list the calling user's AI session backups in the workspace object storage
|
||||
operationId: listAiSessionBackups
|
||||
tags:
|
||||
- ai
|
||||
parameters:
|
||||
- $ref: "#/components/parameters/WorkspaceId"
|
||||
responses:
|
||||
"200":
|
||||
description: backups, newest first; `enabled` is false when the workspace has no storage for them
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
required:
|
||||
- enabled
|
||||
- sessions
|
||||
properties:
|
||||
enabled:
|
||||
type: boolean
|
||||
storage_id:
|
||||
type: string
|
||||
description: names the storage answered from; sync state recorded against another one is void
|
||||
backup_generation:
|
||||
type: integer
|
||||
description: bumped by every workspace key rotation; sync state recorded under another one is void
|
||||
sessions:
|
||||
type: array
|
||||
description: the newest 500 at most
|
||||
items:
|
||||
$ref: "#/components/schemas/AISessionBackupListing"
|
||||
truncated:
|
||||
type: boolean
|
||||
description: the user has more sessions than the answer names
|
||||
|
||||
/w/{workspace}/ai/sessions/pull:
|
||||
post:
|
||||
summary: fetch whole AI session backups
|
||||
operationId: pullAiSessionBackups
|
||||
tags:
|
||||
- ai
|
||||
parameters:
|
||||
- $ref: "#/components/parameters/WorkspaceId"
|
||||
requestBody:
|
||||
required: true
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
required:
|
||||
- ids
|
||||
properties:
|
||||
ids:
|
||||
type: array
|
||||
maxItems: 20
|
||||
items:
|
||||
type: string
|
||||
resume:
|
||||
$ref: "#/components/schemas/AISessionBackupCursor"
|
||||
responses:
|
||||
"200":
|
||||
description: the backups found; `deferred` lists ids that did not fit the response budget
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
required:
|
||||
- enabled
|
||||
- sessions
|
||||
- deferred
|
||||
properties:
|
||||
enabled:
|
||||
type: boolean
|
||||
storage_id:
|
||||
type: string
|
||||
backup_generation:
|
||||
type: integer
|
||||
sessions:
|
||||
type: array
|
||||
items:
|
||||
$ref: "#/components/schemas/AISessionBackup"
|
||||
deferred:
|
||||
type: array
|
||||
items:
|
||||
type: string
|
||||
|
||||
/w/{workspace}/ai/sessions/push:
|
||||
post:
|
||||
summary: write changed pieces of AI sessions to their backups, and remove deleted ones
|
||||
operationId: pushAiSessionBackups
|
||||
tags:
|
||||
- ai
|
||||
parameters:
|
||||
- $ref: "#/components/parameters/WorkspaceId"
|
||||
requestBody:
|
||||
required: true
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
required:
|
||||
- owner
|
||||
properties:
|
||||
owner:
|
||||
type: string
|
||||
description: the email the push was prepared for; refused with a 409 when it is not the caller's
|
||||
sessions:
|
||||
type: array
|
||||
items:
|
||||
$ref: "#/components/schemas/AISessionBackupPush"
|
||||
removed:
|
||||
type: array
|
||||
items:
|
||||
type: string
|
||||
responses:
|
||||
"200":
|
||||
description: one result per session written or removed, in request order
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
required:
|
||||
- enabled
|
||||
- results
|
||||
properties:
|
||||
enabled:
|
||||
type: boolean
|
||||
storage_id:
|
||||
type: string
|
||||
backup_generation:
|
||||
type: integer
|
||||
results:
|
||||
type: array
|
||||
items:
|
||||
type: object
|
||||
required:
|
||||
- id
|
||||
properties:
|
||||
id:
|
||||
type: string
|
||||
error:
|
||||
type: string
|
||||
needs_whole:
|
||||
type: boolean
|
||||
description: nothing was written and the session must be pushed whole again; an incremental part found no listed session to ride on (the backup was removed, or a push split over parts is in progress or was abandoned), or a later part of a push split over parts found another push had superseded it
|
||||
|
||||
/w/{workspace}/ai/shared_artifacts/share:
|
||||
post:
|
||||
summary: share an AI session artifact with the workspace
|
||||
@@ -28278,6 +28425,155 @@ components:
|
||||
fixes) from the workspace UI. Read from the workspace's own settings even when
|
||||
the providers served fall back to the instance config. AI agent steps and the
|
||||
AI sandbox in flows are unaffected.
|
||||
sessions_storage_disabled:
|
||||
type: boolean
|
||||
description: >-
|
||||
Stops browsers from backing their AI sessions up to the workspace's object
|
||||
storage. Read from the workspace's own settings like `copilot_disabled`.
|
||||
|
||||
AISessionBackupListing:
|
||||
type: object
|
||||
required:
|
||||
- id
|
||||
- updated_at
|
||||
- epoch
|
||||
properties:
|
||||
id:
|
||||
type: string
|
||||
updated_at:
|
||||
type: string
|
||||
format: date-time
|
||||
epoch:
|
||||
type: integer
|
||||
description: the session's move count when this copy was pushed; of a session two workspaces list, the copy with the higher one is the later
|
||||
|
||||
AISessionBackupImage:
|
||||
type: object
|
||||
required:
|
||||
- chat_id
|
||||
- id
|
||||
- data_url
|
||||
properties:
|
||||
chat_id:
|
||||
type: string
|
||||
id:
|
||||
type: string
|
||||
data_url:
|
||||
type: string
|
||||
|
||||
AISessionBackupChat:
|
||||
type: object
|
||||
required:
|
||||
- id
|
||||
- record
|
||||
properties:
|
||||
id:
|
||||
type: string
|
||||
record:
|
||||
type: object
|
||||
additionalProperties: true
|
||||
|
||||
AISessionBackup:
|
||||
type: object
|
||||
required:
|
||||
- id
|
||||
- head
|
||||
- chats
|
||||
- images
|
||||
- listing
|
||||
properties:
|
||||
id:
|
||||
type: string
|
||||
head:
|
||||
type: object
|
||||
additionalProperties: true
|
||||
chats:
|
||||
type: array
|
||||
items:
|
||||
$ref: "#/components/schemas/AISessionBackupChat"
|
||||
images:
|
||||
type: array
|
||||
items:
|
||||
$ref: "#/components/schemas/AISessionBackupImage"
|
||||
artifacts:
|
||||
type: object
|
||||
additionalProperties: true
|
||||
next:
|
||||
$ref: "#/components/schemas/AISessionBackupCursor"
|
||||
listing:
|
||||
type: string
|
||||
description: a fingerprint of the session's listing; pages of one session whose fingerprints differ do not belong together
|
||||
moved:
|
||||
type: boolean
|
||||
description: the backup kept changing while this page was read, so it may mix two versions; the browser starts the session over
|
||||
|
||||
AISessionBackupCursor:
|
||||
type: object
|
||||
description: where a pull of a session that did not fit one answer whole picks up; the rest of the session follows a pull naming that session alone with this as `resume`
|
||||
required:
|
||||
- id
|
||||
- images
|
||||
- after
|
||||
properties:
|
||||
id:
|
||||
type: string
|
||||
images:
|
||||
type: boolean
|
||||
after:
|
||||
type: string
|
||||
|
||||
AISessionBackupPush:
|
||||
type: object
|
||||
required:
|
||||
- id
|
||||
properties:
|
||||
id:
|
||||
type: string
|
||||
head:
|
||||
type: object
|
||||
additionalProperties: true
|
||||
chats:
|
||||
type: array
|
||||
items:
|
||||
$ref: "#/components/schemas/AISessionBackupChat"
|
||||
images:
|
||||
type: array
|
||||
items:
|
||||
$ref: "#/components/schemas/AISessionBackupImage"
|
||||
artifacts:
|
||||
type: object
|
||||
additionalProperties: true
|
||||
delete_chats:
|
||||
type: array
|
||||
items:
|
||||
type: string
|
||||
delete_images:
|
||||
type: array
|
||||
items:
|
||||
type: object
|
||||
required:
|
||||
- chat_id
|
||||
- id
|
||||
properties:
|
||||
chat_id:
|
||||
type: string
|
||||
id:
|
||||
type: string
|
||||
partial:
|
||||
type: boolean
|
||||
description: more parts of this session follow, in this push or a later one; the session is not listed on this one. Such a part names its push (`push`), or it is refused
|
||||
whole:
|
||||
type: boolean
|
||||
description: a part of a push of the session whole; the head is on the part that opens it, which replaces whatever the storage holds of the session, and every piece the browser has is on one of them. An incremental part instead rides on a session the storage lists and is refused with needs_whole when it lists none
|
||||
push:
|
||||
type: string
|
||||
description: a push split over several parts names itself on each with a token the browser draws; the part that opens it unlists the session and the last part lists it again, and a later part is written only while that token is the one there (refused with needs_whole otherwise)
|
||||
opens:
|
||||
type: boolean
|
||||
description: this part opens the push named by `push`
|
||||
epoch:
|
||||
type: integer
|
||||
description: the session's move count (its record's `moves`), kept with the marker that lists the session; an incremental part rides on the marker of the same count
|
||||
|
||||
FreeTierInfo:
|
||||
type: object
|
||||
|
||||
@@ -451,6 +451,10 @@ pub struct AIConfig {
|
||||
/// and the AI sandbox are unaffected, so the providers stay in force.
|
||||
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
|
||||
pub copilot_disabled: bool,
|
||||
/// Stops browsers from backing their AI sessions up to the workspace's object storage
|
||||
/// (`ai_sessions.rs`). Read from the workspace's own row like `copilot_disabled`.
|
||||
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
|
||||
pub sessions_storage_disabled: bool,
|
||||
}
|
||||
|
||||
/// Negotiated rates in USD per million tokens. An unset cache rate is read as the
|
||||
@@ -527,6 +531,9 @@ pub fn workspaced_service() -> Router {
|
||||
#[cfg(feature = "bedrock")]
|
||||
let router = router.route("/check_bedrock_credentials", get(check_bedrock_credentials));
|
||||
|
||||
#[cfg(feature = "parquet")]
|
||||
let router = router.nest("/sessions", crate::ai_sessions::workspaced_service());
|
||||
|
||||
router
|
||||
}
|
||||
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -69,6 +69,8 @@ mod ai;
|
||||
#[cfg(feature = "private")]
|
||||
mod ai_free_tier_ee;
|
||||
mod ai_free_tier_oss;
|
||||
#[cfg(feature = "parquet")]
|
||||
mod ai_sessions;
|
||||
mod ai_shared_artifacts;
|
||||
mod apps;
|
||||
mod apps_raw_bundle;
|
||||
|
||||
@@ -147,6 +147,7 @@ async fn edit_copilot_config(
|
||||
|
||||
let workspace_has_config = ai_config.has_providers();
|
||||
let copilot_disabled = ai_config.copilot_disabled;
|
||||
let sessions_storage_disabled = ai_config.sessions_storage_disabled;
|
||||
let instance_ai_config =
|
||||
sqlx::query_scalar!("SELECT value FROM global_settings WHERE name = 'ai_config'")
|
||||
.fetch_optional(&db)
|
||||
@@ -174,6 +175,7 @@ async fn edit_copilot_config(
|
||||
AIConfig::default()
|
||||
};
|
||||
effective_ai_config.copilot_disabled = copilot_disabled;
|
||||
effective_ai_config.sessions_storage_disabled = sessions_storage_disabled;
|
||||
|
||||
Ok(Json(EditCopilotConfigResponse {
|
||||
effective_ai_config,
|
||||
@@ -212,6 +214,9 @@ async fn get_copilot_info(
|
||||
let copilot_disabled = workspace_ai_config
|
||||
.as_ref()
|
||||
.is_some_and(|c| c.0.copilot_disabled);
|
||||
let sessions_storage_disabled = workspace_ai_config
|
||||
.as_ref()
|
||||
.is_some_and(|c| c.0.sessions_storage_disabled);
|
||||
let instance_config =
|
||||
sqlx::query_scalar!("SELECT value FROM global_settings WHERE name = 'ai_config'")
|
||||
.fetch_optional(&db)
|
||||
@@ -236,6 +241,7 @@ async fn get_copilot_info(
|
||||
AIConfig::default()
|
||||
};
|
||||
effective.copilot_disabled = copilot_disabled;
|
||||
effective.sessions_storage_disabled = sessions_storage_disabled;
|
||||
Ok(Json(effective))
|
||||
}
|
||||
|
||||
|
||||
@@ -252,12 +252,17 @@ pub async fn build_crypt_with_key_suffix(
|
||||
key_suffix: &str,
|
||||
) -> crate::error::Result<MagicCrypt256> {
|
||||
let key = get_workspace_key(w_id, db).await?;
|
||||
Ok(crypt_from_key_with_suffix(&key, key_suffix))
|
||||
}
|
||||
|
||||
/// The cipher `build_crypt_with_key_suffix` builds, from a key string in hand.
|
||||
pub fn crypt_from_key_with_suffix(key: &str, key_suffix: &str) -> MagicCrypt256 {
|
||||
let crypt_key = if let Some(ref salt) = SECRET_SALT.as_ref() {
|
||||
format!("{}{}{}", key, salt, key_suffix)
|
||||
} else {
|
||||
format!("{}{}", key, key_suffix)
|
||||
};
|
||||
Ok(magic_crypt::new_magic_crypt!(crypt_key, 256))
|
||||
magic_crypt::new_magic_crypt!(crypt_key, 256)
|
||||
}
|
||||
|
||||
pub async fn get_workspace_key(w_id: &str, db: &DB) -> crate::error::Result<String> {
|
||||
|
||||
@@ -68,8 +68,8 @@ pub mod object_store_reexports {
|
||||
pub use object_store::path::Path;
|
||||
pub use object_store::{
|
||||
Attribute, Attributes, Error as ObjectStoreError, GetOptions, GetRange, GetResult,
|
||||
ObjectStore, PutMultipartOpts, PutPayload, PutResult, Result as ObjectStoreResult,
|
||||
WriteMultipart,
|
||||
ObjectMeta, ObjectStore, PutMode, PutMultipartOpts, PutOptions, PutPayload, PutResult,
|
||||
Result as ObjectStoreResult, UpdateVersion, WriteMultipart,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,266 @@
|
||||
# AI session backups
|
||||
|
||||
AI sessions live in the browser: the session list (`windmill-sessions`), chat transcripts and
|
||||
image blobs (`copilot-chat-history`) and artifacts (`copilot-artifacts`), all per-user IndexedDB
|
||||
stores. This is the design of their backup in the workspace's object storage, and the
|
||||
constraints future work on either side must keep.
|
||||
|
||||
Backend: `backend/windmill-api/src/ai_sessions.rs` (`/w/{w}/ai/sessions/{list,pull,push}`).
|
||||
Frontend: `frontend/src/lib/components/sessions/sessionMirror*.ts`.
|
||||
|
||||
## Why it is lazy
|
||||
|
||||
A session changes at the local write rate: a transcript write every 2 s while streaming, a
|
||||
session-record write per new message on screen. An object in S3 is replaced whole and every PUT
|
||||
is billed, so the backup deliberately does not follow that rate. Local writes only mark a session
|
||||
dirty (`sessionMirrorSignal.ts`, import-free so the stores never depend on the backup). A flush
|
||||
runs 15 s after the marks go quiet, at most 2 min after the first unflushed mark, when the tab is
|
||||
hidden, and 10 s after load for marks a crash left behind. Marks are persisted in localStorage
|
||||
(shared by the user's tabs) for that reason, one key per mark: a shared blob would let two tabs
|
||||
marking different sessions at once rewrite each other's mark away. A dirty mark is a counter
|
||||
bumped on every write; a push retires it by recording the counter it covered on the session's
|
||||
sync row rather than deleting the mark, since two localStorage calls cannot compare-and-delete
|
||||
and a bump landing between them would be lost; retired marks are not reclaimed (one small key per
|
||||
session ever backed up), and the marks of unsent drafts and of workspaces that are off stay too,
|
||||
each costing one lookup per flush. Only a session gone from the store has its mark deleted. Losing the last
|
||||
seconds of a device that never comes back is accepted; a tab that closes normally keeps its marks.
|
||||
A signal names the user whose store the write landed in (read off the store's scoped name), so
|
||||
a write that completes after the logged-in user changed marks that user's session, for their
|
||||
next load, rather than the current user's.
|
||||
|
||||
A flush plans and sends one session at a time, filling requests of about 8 MB as it goes, so a
|
||||
first backfill of a large history never holds more than one request's worth of records and
|
||||
images in memory.
|
||||
|
||||
## What a push carries
|
||||
|
||||
The pure planner (`sessionMirrorPlan.ts`) compares each piece against the marker of what was
|
||||
last pushed, kept per session in the `windmill-sessions-mirror` store:
|
||||
|
||||
| Piece | Object | Sent when |
|
||||
|---|---|---|
|
||||
| session record | `sessions/{sid}/head.json` | its signature changed |
|
||||
| chat | `sessions/{sid}/chats/{cid}.json` | its `lastModified` moved |
|
||||
| artifacts | `sessions/{sid}/artifacts.json` | their fingerprint changed |
|
||||
| image | `images/{sid}/{cid}/{iid}` | never pushed before (write-once) |
|
||||
| index marker | `index/{sid}/{epoch}` | last, by the part that completes a push of the session (empty; named by the record's move count) |
|
||||
|
||||
All under `windmill_ai_sessions/{w_id}/g{generation}/{sha256(email)}/` in the workspace's
|
||||
primary storage (the generation is what a key rotation moves, see below).
|
||||
The listing reads only `index/`: one object per session whatever the session holds, so a
|
||||
session with many chats cannot crowd newer ones out of a bounded scan, and its
|
||||
`last_modified` is the session's `updated_at`. Written last, and only by an entry no unsent
|
||||
part follows (a session split over several entries says `partial` on all but the last, and
|
||||
names the push on each, or the part is refused), it
|
||||
lists a session only once a whole push landed; the parts of a session after a failed one are
|
||||
not written either, on the server within one push and on the client across pushes, so the
|
||||
marker on the last part never lists a session missing a chat, and a new session whose last part
|
||||
never lands is not listed at all. A push of the session whole (no sync row, or a stale one,
|
||||
`whole` on every part) opens with the head on the first: that part replaces the backup (the
|
||||
marker goes first, then everything under the session), so what an old storage still held of
|
||||
the session and the push does not carry is gone. An incremental part rides on a listed
|
||||
session, and the server refuses it with `needs_whole`, writing nothing, when none is listed
|
||||
(a removal deletes the marker first), rather than write a marker over a session missing what
|
||||
earlier parts or earlier pushes carried; one that changes more than one object unlists the
|
||||
session before its writes and lists it again after them, so a pull between two of the writes,
|
||||
or after one failed, finds it absent rather than a mix of old and new pieces (one object
|
||||
changing is one write, and stays listed). A push split over several parts, whole or
|
||||
incremental, names itself on each with a token the browser draws (`push`, `opens` on the
|
||||
first): the opening part unlists the session, so a pull between two parts finds it absent
|
||||
rather than a mix of old and new pieces, the last part lists it again, and a later part is
|
||||
written only while that token is the one there, so two devices pushing the session at once
|
||||
cannot list a mix of their pieces (the push that opened later wins; the other is refused with
|
||||
`needs_whole` and goes again), and one abandoned leaves the session unlisted, so the next
|
||||
push of it goes whole. A push and a removal of one session
|
||||
are serialized on the server by a Postgres advisory lock keyed on the session's prefix, so
|
||||
the two never interleave object by object.
|
||||
|
||||
The head signature leaves out `name` (a per-browser counter the sessions page routes by),
|
||||
the unsent-draft fields, `workspace_root_id` (recomputed on import), and the two fields reading
|
||||
a session bumps (`lastSeenCount`, `lastActivityAt`). Reading a session must never cost a push;
|
||||
keep that property when adding fields to `Session`.
|
||||
|
||||
Unsent drafts (no `workspace_id`) and attached files (Blobs, directory handles) are not backed up.
|
||||
|
||||
## Encryption and access
|
||||
|
||||
Every object is encrypted with a key derived from the workspace key and the user
|
||||
(`build_crypt_with_key_suffix` with the email hash), because workspace storage credentials are
|
||||
shared far more widely than a user's transcripts: `public_resource` storages and legacy-mode
|
||||
READ/WRITE hand any member the bucket. The key is per user rather than per workspace so that a
|
||||
member who copies another user's ciphertext under their own prefix gets nothing from `pull`; an
|
||||
object that does not decrypt for its reader is treated as absent. Rotating the workspace key
|
||||
(`set_encryption_key`) does not re-key the backups the way it re-encrypts the workspace's
|
||||
secrets. The objects live under a prefix named by a generation
|
||||
(`workspace_settings.ai_sessions_backup_generation`) that the rotation bumps in the
|
||||
transaction committing the new key; once committed, the routes read and write under the new
|
||||
generation's prefix, the answers name it (`backup_generation`, below; `storage_id` names the
|
||||
storage and does not change), so every browser marks its sync rows stale and pushes its
|
||||
sessions whole again there, and every older generation, which nothing writes to any more, is deleted off the
|
||||
request at leisure (`windmill-api-workspaces/src/ai_session_backups.rs`). Sessions no browser
|
||||
holds any more are lost. A generation is never reused, so no deletion, however late, can
|
||||
touch live objects; a rotation that fails before its commit bumps nothing and deletes
|
||||
nothing; two rotations racing serialize on the key row; the same key set again bumps
|
||||
nothing. A rotation is rare, and the alternative,
|
||||
rewriting every object in place while pushes, restarts, storage switches and further
|
||||
rotations race the rewrite, is where the complexity would be; with this, nothing but the
|
||||
current key ever reads an object. The
|
||||
server builds every key from ids it validated
|
||||
(`[A-Za-z0-9_-]{1,64}`) and the caller's own email; the client never names a key, and the
|
||||
workspace storage permission rules are not consulted (the same stance as volumes). Only an
|
||||
unscoped user token may reach the routes: a job token can carry an `on_behalf_of` identity and
|
||||
every scoped token (guest, embed, app policy, MCP) was minted for something narrower.
|
||||
|
||||
The backup is keyed by the email like the browser's own stores are (`userScopedDb` scopes
|
||||
IndexedDB by it): a user whose email changes starts from an empty history on both sides, and
|
||||
the objects under the old hash stay in the bucket unread. Carrying them over would need a
|
||||
server-side re-key (decrypt with the old suffix, encrypt with the new, move every object) in the
|
||||
email-change flow, which this design leaves out.
|
||||
|
||||
An image is accepted only as a base64 data URL of at most 4 MB and stored verbatim, so it
|
||||
serializes back into a pull answer at its stored size; anything JSON would escape could grow
|
||||
several times and defeat the pull budget.
|
||||
|
||||
`push` carries `owner`, the email the browser prepared the batch for, and the server refuses a
|
||||
mismatch with 409: an in-place account switch must not file one user's sessions under another's
|
||||
prefix. The client captures its user at flush start and checks every store handle's name
|
||||
against it for the same reason.
|
||||
|
||||
The feature is on wherever the workspace has primary storage, and off with
|
||||
`ai_config.sessions_storage_disabled` (the `copilot_disabled` pattern: no migration, carried by
|
||||
settings export and the CLI). A build without `parquet` has no routes (404), a workspace without
|
||||
storage answers `enabled: false`; either turns the backup off for ten minutes, after which the
|
||||
page asks again on its own (a flush for whatever is pending, and a restore), and the AI
|
||||
settings page tells the mirror at once when the switch is saved there (the off state is
|
||||
forgotten, the rows that went stale are marked again, a restore runs).
|
||||
|
||||
## Conflicts and deletion
|
||||
|
||||
Last write wins across devices. The head carries no manifest; `pull` lists the session's prefix
|
||||
instead, so a stale device that renames or archives a session rewrites only the head and cannot
|
||||
hide chats a newer device wrote. Two devices continuing the same chat still collide.
|
||||
|
||||
Restore brings back only sessions the browser does not have (`importSessions` is write-if-absent,
|
||||
and skips ids the user deleted in this page) and never overwrites or deletes a local one from
|
||||
remote state. It covers the workspace and its forks together, and only once every one of them
|
||||
that keeps backups has listed (a listing that failed leaves the family for the next page load
|
||||
or workspace switch, or the copy that did list could be the stale one): a session listed by
|
||||
two of them (moved between them, the old copy not yet removed, since that mark is the moving
|
||||
browser's, which may never come back) is brought back from the copy that moved last (`epoch`,
|
||||
the record's move count, which names the marker), the storage's own modification time
|
||||
deciding between two of the same count, and not from the other, which would otherwise take
|
||||
the id first and keep the later copy out for good. A workspace's records land together once
|
||||
its pulls are done, and just before they do the whole family is listed again (members whose
|
||||
backups were off included, since a move from another device can land in a workspace between
|
||||
the first listings and the pulls; a family of one, with nowhere else for a copy to show up, is
|
||||
not): a session a later copy of which showed up elsewhere is left, with the family, for the
|
||||
next time. Only a user-initiated `deleteSession` removes the backup; the next push from
|
||||
another device that still has the session is refused with `needs_whole` (nothing of it is
|
||||
written), its row goes stale without a backoff, and that device's next flush sends the session
|
||||
whole; the workspace-lifecycle
|
||||
removals (`reconcileSessionsLifecycle`, `deleteSessionsForWorkspace`) leave it, so a session
|
||||
dropped by a wrong reconcile comes back on the next restore. Objects of deleted workspaces stay
|
||||
in the bucket. A session moved to another workspace is pushed whole into the new one, and once
|
||||
that push has landed the copy in the old one gets a removal mark of its own, naming the storages holding
|
||||
that copy (the row that knew is the new workspace's by then), retried independently until
|
||||
each of them has answered, even when the old workspace's backups are off at the time (they
|
||||
may hold the copy still). Filing the removal only after the new copy is acknowledged keeps the
|
||||
session backed up somewhere at every point.
|
||||
|
||||
A restore writes a session's artifacts and chats before its record, and records nothing for a
|
||||
session whose pieces could not be written: recording it would let the next flush push the
|
||||
half-empty local state over the backup.
|
||||
|
||||
Every answer names the storage it came from (`storage_id`, a hash of what locates the objects,
|
||||
endpoint, region and bucket, not the credentials, which rotate) and the backup generation a
|
||||
key rotation bumps (`backup_generation`). A sync row records both, and a row naming another
|
||||
storage or generation goes stale and its session is marked again: a workspace pointed at a
|
||||
new bucket, or whose key was rotated, holds nothing, and the server looks nowhere else, so
|
||||
the next flush carries the session whole. A switch leaves the old copy where it was, so the
|
||||
row rewritten under the new storage records the old one (`alsoIn`, one entry per storage
|
||||
the workspace was on), and a removal is done only once every storage holding a copy answered
|
||||
it, whatever the generation (a rotation deleted the older generation's copy anyway): each
|
||||
answer narrows the row to the storages still holding one, and the mark waits for them to
|
||||
answer, so a switch back never brings a deleted session back. That includes the rows a flush has just written, when a later answer of the
|
||||
same flush names another storage or the session was pushed in part on top of a row from the
|
||||
old one; a session whose own parts were answered from different storages is not settled at
|
||||
all. The listing a restore starts with runs the same check, so a storage switch is noticed at
|
||||
the first push after it or on the next page load, whichever comes first.
|
||||
|
||||
## Limits
|
||||
|
||||
Push bodies are packed to about 8 MB (UTF-8 bytes as sent), at most 100 entries, 200 removals and
|
||||
4000 pieces each (the server's caps, with 32 MB on the body, and 100 chats, 500 images or 1000
|
||||
deletes per entry, since every piece is an object-store call); an entry that
|
||||
outgrows the target is split into chat-only parts (the artifacts and deletes on the last, the
|
||||
head on the last too for an incremental push and on the first part, whatever it carries, for a
|
||||
push of the session whole), and deletes
|
||||
past the per-entry cap are carried over to the next push, which the session stays marked for. A chat above
|
||||
16 MB or a session's artifacts above 8 MB are left out with a console warning; a chat that grew
|
||||
past the cap after it was backed up has its copy deleted, so a restore never presents the old
|
||||
transcript as the current one. A 413 fails only the sessions of that request. A
|
||||
request the server refuses (any other 4xx but 404/403/409) stops the backup for the page but keeps
|
||||
the marks and the sync state, so the next load tries again; a session the server reports it could
|
||||
not store stays marked and is retried with backoff. A move files the old workspace's removal
|
||||
mark before recording the new copy's row, so a mark that could not be written leaves the move
|
||||
to be planned again. A workspace that answers `enabled: false`
|
||||
marks its sync rows stale (the next push after storage returns carries every session whole,
|
||||
since a new storage may be a new bucket) and leaves its dirty marks where they are (a move into
|
||||
it must still remember the old copy); it keeps the removal marks of sessions that had been backed
|
||||
up, so one deleted while backups are off does not come back once they are on, and drops the
|
||||
removals of sessions never backed up from this browser, so a storage-less instance does not
|
||||
collect one mark per deleted session forever. A user delete whose removal mark cannot be written
|
||||
(localStorage full) is carried by the session's sync row instead (`removed`), which the flush
|
||||
and the restore read like a mark; a session without a row yet (its first push may be in
|
||||
flight) gets a row saying only that, and every row write keeps a removal filed meanwhile, so
|
||||
the push's own row cannot erase it. Pull bodies are
|
||||
capped at 64 KB. Pull answers up to 20 ids within a 32 MB
|
||||
budget: a session's size is known from the listings before anything of it is read, one that
|
||||
would not fit is deferred unless it is the first of the answer, in which case it comes in
|
||||
pages: the answer carries what fits in key order (at least one object, so every page makes
|
||||
progress) and names where the next picks up (`next`, a cursor the browser sends back as
|
||||
`resume` with that session alone). Every page carries a fingerprint of the session's listing
|
||||
(marker, keys, sizes, modification times, entity tags and versions, since a store reports
|
||||
modification times coarsely and an object rewritten at the same size within that grain would
|
||||
otherwise fingerprint the same) taken before anything of it is read, and the server
|
||||
takes it again once the page is read: a page the backup moved under (a push landing object by
|
||||
object) is read again, a few times, then answered as `moved`, and the browser starts the
|
||||
session over on that or on two pages whose fingerprints differ. The browser writes each page's pieces as it arrives, over whatever
|
||||
an earlier restore cut short had staged (the session is absent locally, so its pieces have no
|
||||
local edits to keep, and the backup may have moved on), and the record, which is what makes
|
||||
the session visible, only with the last page. A restore in progress keeps a staging row for
|
||||
the session (the ids of every chat, image, artifact and version it wrote), which outlives it
|
||||
if it is cut short; the next restore deletes the staged pieces the backup no longer has, by id
|
||||
and never by clock, before the record lands (once the record is there no restore looks at the
|
||||
session again, and a flush would push them back), and a prune that could not run leaves the
|
||||
session, its pieces and its staging row for the restore after. A restore holds the user's tab
|
||||
lock while it runs, so two tabs cannot each write the same absent session's pieces over the
|
||||
other's, and runs only where Web Locks exist (a secure context: https, or localhost); on a plain
|
||||
http origin the browser still backs up, and its sessions come back on a secure one. A
|
||||
restore never writes an older record over a newer
|
||||
one; between pages it holds nothing but the sync
|
||||
row being assembled, whose chats also admit the images of a later page. Every page carries a
|
||||
fingerprint of the session's listing (`listing`), taken before anything of the page is
|
||||
listed or read, so an object landing after it is in the next page's; a session whose
|
||||
fingerprint moved between two of its pages (a chat added by another device could sort before
|
||||
the cursor and be missed) starts over, up to three times, then waits for the next restore. An object that grew
|
||||
since the listing (a push replaced it) ends its page just before it and the answer names that
|
||||
spot, so the next page sizes it anew rather than the session being imported without it. A pull sees every key of a session's listing but keeps the 5000 smallest
|
||||
past its cursor (a page is defined by key order, and the store promises none), so a session
|
||||
grown without bound by valid pushes cannot grow the answer's memory through its metadata
|
||||
either; removing a prefix and a rotation's deletion stream their listings. `list` scans at most 50 000 index markers, keeps the newest 500 as it goes and answers with
|
||||
them (`truncated` says when there were more); the restore takes 50 of them. Every read checks the object's size before buffering it: one larger than any push writes
|
||||
(32 MB) is planted, whatever its listing said, and skipped, since whoever holds the bucket's
|
||||
credentials can put anything at a predictable key; one larger than its listing said grew
|
||||
since (a push replaced it) and ends its page, for the next pull to size anew. A dirty mark that cannot be written
|
||||
(localStorage full) records its bump on the session's sync row instead (`extraV`, counted
|
||||
with the mark's counter and kept by every row write, so a push in flight cannot retire it);
|
||||
a session without a row yet (its first push in flight) keeps the bump in the page, and the
|
||||
next row write takes it onto the row in the same transaction, so the row the push writes
|
||||
cannot retire the mark with the bump unseen; every load's backfill marks again a session
|
||||
without a row, with a stale one, or with one carrying bumps no push has covered.
|
||||
A mark or removal for another user (a write that landed after a switch) reaches that user's
|
||||
rows through a connection of its own, since the shared handle follows the current user. Nothing is read past the budget, whatever a session holds. A
|
||||
restore takes the newest 50 sessions per workspace: every visible session gets a runtime, and
|
||||
each runtime's history load reads the whole chat store. On CE the push checks the storage quota
|
||||
and bumps usage by bytes written (an over-count on overwrites; the periodic recount settles it).
|
||||
@@ -1084,12 +1084,13 @@
|
||||
it with none, whether SSO logins evaluate an IdP groups claim (SAML or OIDC) and
|
||||
change a membership, the plan tier and quota shown when the execution meter is opened,
|
||||
whether app sandbox isolation is turned on, whether a step's workspace script is
|
||||
edited from the flow editor, which skin approval steps are given, whether AI agent
|
||||
steps set their own memory id as a fixed id or an expression, how data tables and
|
||||
their migrations are set up and used, how often an empty workspace home is seen, how
|
||||
often the home page’s create menu and hub-project picker are opened and from which
|
||||
entry point, the name of any public hub project imported from the home page and how
|
||||
far that import got, and whether a pre-approved trial offer was opened, last 30 days)</li
|
||||
edited from the flow editor, which skin approval steps are given, how many AI sessions
|
||||
are brought back from the workspace object storage backup, whether AI agent steps set
|
||||
their own memory id as a fixed id or an expression, how data tables and their
|
||||
migrations are set up and used, how often an empty workspace home is seen, how often
|
||||
the home page’s create menu and hub-project picker are opened and from which entry
|
||||
point, the name of any public hub project imported from the home page and how far that
|
||||
import got, and whether a pre-approved trial offer was opened, last 30 days)</li
|
||||
>
|
||||
<li
|
||||
>feature adoption (counts of which flow, script, trigger, worker and data table
|
||||
@@ -1150,12 +1151,13 @@
|
||||
it with none, whether SSO logins evaluate an IdP groups claim (SAML or OIDC) and
|
||||
change a membership, the plan tier and quota shown when the execution meter is opened,
|
||||
whether app sandbox isolation is turned on, whether a step's workspace script is
|
||||
edited from the flow editor, which skin approval steps are given, whether AI agent
|
||||
steps set their own memory id as a fixed id or an expression, how data tables and
|
||||
their migrations are set up and used, how often an empty workspace home is seen, how
|
||||
often the home page’s create menu and hub-project picker are opened and from which
|
||||
entry point, the name of any public hub project imported from the home page and how
|
||||
far that import got, and whether a pre-approved trial offer was opened, last 30 days)</li
|
||||
edited from the flow editor, which skin approval steps are given, how many AI sessions
|
||||
are brought back from the workspace object storage backup, whether AI agent steps set
|
||||
their own memory id as a fixed id or an expression, how data tables and their
|
||||
migrations are set up and used, how often an empty workspace home is seen, how often
|
||||
the home page’s create menu and hub-project picker are opened and from which entry
|
||||
point, the name of any public hub project imported from the home page and how far that
|
||||
import got, and whether a pre-approved trial offer was opened, last 30 days)</li
|
||||
>
|
||||
<li
|
||||
>feature adoption (counts of which flow, script, trigger, worker and data table
|
||||
|
||||
@@ -1,9 +1,15 @@
|
||||
import { type DBSchema as IDBSchema, type IDBPDatabase } from 'idb'
|
||||
import {
|
||||
type DBSchema as IDBSchema,
|
||||
type IDBPDatabase,
|
||||
type IDBPTransaction,
|
||||
type StoreNames
|
||||
} from 'idb'
|
||||
import type { ChatJob, DisplayMessage } from './shared'
|
||||
import { expanded, messageDraft } from './chatDraft'
|
||||
import { createLongHash } from '$lib/editorLangUtils'
|
||||
import { userScopedDb, type UserScopedDbMigrateDeps } from '$lib/userScopedDb'
|
||||
import { scopedKey } from '$lib/userScopedStorage'
|
||||
import { emailOfScopedKey, scopedKey, scopedKeyFor } from '$lib/userScopedStorage'
|
||||
import { markSessionDirty } from '$lib/components/sessions/sessionMirrorSignal'
|
||||
import type { ChatCompletionMessageParam } from 'openai/resources/index.mjs'
|
||||
import type { PersistedContextUsage } from './tokenUsage'
|
||||
import { IMAGE_OMITTED_PLACEHOLDER, type AttachedImage } from './imageUtils'
|
||||
@@ -14,8 +20,9 @@ import { randomUUID } from '$lib/utils/uuid'
|
||||
// shared browser. The bare name is also the legacy (pre-namespacing) DB, claimed
|
||||
// once on first login.
|
||||
const DB_NAME = 'copilot-chat-history'
|
||||
// v3 adds the images blob store (replacing v2's short-lived toolImages store).
|
||||
const DB_VERSION = 3
|
||||
// v3 adds the images blob store (replacing v2's short-lived toolImages store); v4 indexes
|
||||
// chats by session.
|
||||
const DB_VERSION = 4
|
||||
/** Newest image blobs kept per chat; each is a bounded (≤1568px) data URL. */
|
||||
const MAX_IMAGES_PER_CHAT = 30
|
||||
/** Marks a persisted image whose bytes live in the `images` store. */
|
||||
@@ -45,6 +52,7 @@ interface ChatSchema extends IDBSchema {
|
||||
// chats predating this feature. Persisted out-of-band like modifiedItems.
|
||||
backgroundJobs?: ChatJob[]
|
||||
}
|
||||
indexes: { 'by-session': string }
|
||||
}
|
||||
// Image bytes, out-of-band from the chat record on purpose: the record is
|
||||
// re-cloned into IndexedDB on every saveChat, while a blob is written once
|
||||
@@ -64,9 +72,19 @@ interface ChatSchema extends IDBSchema {
|
||||
}
|
||||
}
|
||||
|
||||
function createChatStore(db: IDBPDatabase<ChatSchema>): void {
|
||||
if (!db.objectStoreNames.contains('chats')) {
|
||||
db.createObjectStore('chats', { keyPath: 'id' })
|
||||
/** A persisted chat, exactly as the store holds it (image refs, not bytes). */
|
||||
export type StoredChat = ChatSchema['chats']['value']
|
||||
|
||||
function createChatStore(
|
||||
db: IDBPDatabase<ChatSchema>,
|
||||
tx: IDBPTransaction<ChatSchema, StoreNames<ChatSchema>[], 'versionchange'>
|
||||
): void {
|
||||
const chats = db.objectStoreNames.contains('chats')
|
||||
? tx.objectStore('chats')
|
||||
: db.createObjectStore('chats', { keyPath: 'id' })
|
||||
// Lets the session backup find a session's chats without reading every record.
|
||||
if (!chats.indexNames.contains('by-session')) {
|
||||
chats.createIndex('by-session', 'sessionId')
|
||||
}
|
||||
// v2 briefly kept full-resolution tool screenshots in their own store; the
|
||||
// general blob store below covers them now.
|
||||
@@ -117,7 +135,9 @@ async function claimLegacyChatDb(
|
||||
{ openDB, deleteDB }: UserScopedDbMigrateDeps
|
||||
): Promise<void> {
|
||||
if ((await scopedDb.count('chats')) > 0) return
|
||||
const legacy = await openDB<ChatSchema>(DB_NAME, 1, { upgrade: createChatStore })
|
||||
const legacy = await openDB<ChatSchema>(DB_NAME, 1, {
|
||||
upgrade: (db, _oldVersion, _newVersion, tx) => createChatStore(db, tx)
|
||||
})
|
||||
const legacyChats = await legacy.getAll('chats')
|
||||
if (legacyChats.length > 0) {
|
||||
const tx = scopedDb.transaction('chats', 'readwrite')
|
||||
@@ -156,6 +176,129 @@ export async function readChatModifiedItems(chatId: string): Promise<string[] |
|
||||
}
|
||||
}
|
||||
|
||||
// Store access for the session backup, outside any manager: it runs for sessions that have
|
||||
// no runtime mounted. Every call names the user it works for and gets nothing back once
|
||||
// the logged-in user differs, so a flush prepared for one user never reads or writes the
|
||||
// next user's history.
|
||||
const backupDbh = userScopedDb<ChatSchema>(DB_NAME, {
|
||||
version: DB_VERSION,
|
||||
upgrade: createChatStore,
|
||||
migrate: migrateLegacyChatDb
|
||||
})
|
||||
|
||||
async function backupDb(email: string): Promise<IDBPDatabase<ChatSchema> | undefined> {
|
||||
const db = await backupDbh.whenReady()
|
||||
return db && db.name === scopedKeyFor(DB_NAME, email) ? db : undefined
|
||||
}
|
||||
|
||||
/** Test-only: let go of the backup's handle so the next call opens the test's fresh
|
||||
* IndexedDB rather than the connection a previous test left. */
|
||||
export function __resetBackupStoreForTesting(): void {
|
||||
backupDbh.close()
|
||||
}
|
||||
|
||||
/** Ids of the chats tagged with this session, or undefined when the store is unavailable. */
|
||||
export async function listSessionChatIds(
|
||||
sessionId: string,
|
||||
email: string
|
||||
): Promise<string[] | undefined> {
|
||||
const db = await backupDb(email)
|
||||
if (!db) return undefined
|
||||
return (await db.getAllKeysFromIndex('chats', 'by-session', sessionId)).map(String)
|
||||
}
|
||||
|
||||
export async function readStoredChat(id: string, email: string): Promise<StoredChat | undefined> {
|
||||
const db = await backupDb(email)
|
||||
return db?.get('chats', id)
|
||||
}
|
||||
|
||||
/** Ids of the image blobs a chat owns, or undefined when the store is unavailable. */
|
||||
export async function listChatImageIds(
|
||||
chatId: string,
|
||||
email: string
|
||||
): Promise<string[] | undefined> {
|
||||
const db = await backupDb(email)
|
||||
if (!db) return undefined
|
||||
return (await imageKeysForChat(db, chatId)).map(String)
|
||||
}
|
||||
|
||||
export async function readImageDataUrl(id: string, email: string): Promise<string | undefined> {
|
||||
const db = await backupDb(email)
|
||||
return (await db?.get('images', id))?.dataUrl
|
||||
}
|
||||
|
||||
export interface RestoredImage {
|
||||
id: string
|
||||
chatId: string
|
||||
dataUrl: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Write restored chats and image blobs, leaving any that already exist alone: a record
|
||||
* this browser wrote since is newer than the backup it came from. False when the store
|
||||
* could not be reached, which the caller must not record as a restore.
|
||||
*/
|
||||
export async function importStoredChats(
|
||||
chats: StoredChat[],
|
||||
images: RestoredImage[],
|
||||
email: string,
|
||||
overwrite = false
|
||||
): Promise<boolean> {
|
||||
const db = await backupDb(email)
|
||||
if (!db) return false
|
||||
const tx = db.transaction(['chats', 'images'], 'readwrite')
|
||||
const chatStore = tx.objectStore('chats')
|
||||
const imageStore = tx.objectStore('images')
|
||||
const savedAt = Date.now()
|
||||
for (const image of images) {
|
||||
if (overwrite || (await imageStore.getKey(image.id)) === undefined) {
|
||||
await imageStore.put({ id: image.id, chatId: image.chatId, dataUrl: image.dataUrl, savedAt })
|
||||
}
|
||||
}
|
||||
// An overwrite never puts an older record over a newer one: without a cross-tab lock,
|
||||
// another restore may have landed a newer backup's copy meanwhile.
|
||||
for (const chat of chats) {
|
||||
const existing = await chatStore.get(chat.id)
|
||||
if (existing === undefined || (overwrite && existing.lastModified <= chat.lastModified)) {
|
||||
await chatStore.put(chat)
|
||||
}
|
||||
}
|
||||
await tx.done
|
||||
return true
|
||||
}
|
||||
|
||||
/** Deletes these chats of the session (with their images) and these images: what an earlier
|
||||
* restore staged for it and the backup no longer has. False when nothing could be deleted. */
|
||||
export async function pruneSessionChats(
|
||||
sessionId: string,
|
||||
chats: Set<string>,
|
||||
images: Set<string>,
|
||||
email: string
|
||||
): Promise<boolean> {
|
||||
if (chats.size === 0 && images.size === 0) return true
|
||||
const db = await backupDb(email)
|
||||
if (!db) return false
|
||||
try {
|
||||
const tx = db.transaction(['chats', 'images'], 'readwrite')
|
||||
const chatStore = tx.objectStore('chats')
|
||||
const imageStore = tx.objectStore('images')
|
||||
for (const chatId of await chatStore.index('by-session').getAllKeys(sessionId)) {
|
||||
if (!chats.has(String(chatId))) continue
|
||||
await chatStore.delete(chatId)
|
||||
const keys = await imageStore
|
||||
.index('by-chat')
|
||||
.getAllKeys(IDBKeyRange.bound([chatId, -Infinity], [chatId, Infinity]))
|
||||
for (const key of keys) await imageStore.delete(key)
|
||||
}
|
||||
for (const id of images) await imageStore.delete(id)
|
||||
await tx.done
|
||||
return true
|
||||
} catch (err) {
|
||||
console.error('Could not prune chats for session', err)
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
export default class HistoryManager {
|
||||
// Per-instance handle to the shared per-user DB lifecycle. There is one
|
||||
// HistoryManager per AIChatManager (the singleton + one per session runtime),
|
||||
@@ -289,7 +432,10 @@ export default class HistoryManager {
|
||||
const snapshot = $state.snapshot(existing)
|
||||
const updated = { ...snapshot, sessionId }
|
||||
this.savedChats = { ...this.savedChats, [chatId]: updated }
|
||||
await this.enqueueDbWrite((db) => db.put('chats', updated))
|
||||
await this.enqueueDbWrite(async (db) => {
|
||||
await db.put('chats', updated)
|
||||
markSessionDirty(sessionId, chatId, emailOfScopedKey(DB_NAME, db.name))
|
||||
})
|
||||
}
|
||||
|
||||
getPastChats() {
|
||||
@@ -565,6 +711,13 @@ export default class HistoryManager {
|
||||
const keep = this.keptImageIds(refs)
|
||||
await this.writeKeptImageBlobs(db, updatedChat.id, blobs, keep)
|
||||
await db.put('chats', updatedChat)
|
||||
if (updatedChat.sessionId) {
|
||||
markSessionDirty(
|
||||
updatedChat.sessionId,
|
||||
updatedChat.id,
|
||||
emailOfScopedKey(DB_NAME, db.name)
|
||||
)
|
||||
}
|
||||
// Best-effort: the record is already committed, so a failed cleanup
|
||||
// (e.g. a user switch closed this handle mid-op) must not turn a
|
||||
// successful save into a rejection — the orphans are reclaimed by
|
||||
@@ -589,6 +742,7 @@ export default class HistoryManager {
|
||||
}
|
||||
|
||||
deletePastChat(id: string) {
|
||||
const sessionId = this.savedChats[id]?.sessionId
|
||||
this.savedChats = Object.fromEntries(
|
||||
Object.entries(this.savedChats).filter(([key]) => key !== id)
|
||||
)
|
||||
@@ -596,6 +750,7 @@ export default class HistoryManager {
|
||||
await db.delete('chats', id)
|
||||
const keys = await imageKeysForChat(db, id)
|
||||
await Promise.all(keys.map((key) => db.delete('images', key)))
|
||||
if (sessionId) markSessionDirty(sessionId, id, emailOfScopedKey(DB_NAME, db.name))
|
||||
}).catch((err) => console.error('Could not delete chat', err))
|
||||
}
|
||||
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
// active chat's rotation, so chatId-keying would drop artifacts on each new conversation.
|
||||
import { type DBSchema as IDBSchema, type IDBPObjectStore, type IDBPTransaction } from 'idb'
|
||||
import { userScopedDb } from '$lib/userScopedDb'
|
||||
import { emailOfScopedKey, scopedKeyFor } from '$lib/userScopedStorage'
|
||||
import { markSessionDirty } from '$lib/components/sessions/sessionMirrorSignal'
|
||||
|
||||
export type ArtifactKind = 'md' | 'html'
|
||||
|
||||
@@ -89,9 +91,11 @@ interface ArtifactsSchema extends IDBSchema {
|
||||
}
|
||||
}
|
||||
|
||||
const ARTIFACTS_DB = 'copilot-artifacts'
|
||||
|
||||
// User-scoped like the chat-history store these are keyed against: no cross-user
|
||||
// co-residency on a shared browser.
|
||||
const dbh = userScopedDb<ArtifactsSchema>('copilot-artifacts', {
|
||||
const dbh = userScopedDb<ArtifactsSchema>(ARTIFACTS_DB, {
|
||||
version: 2,
|
||||
// Runs for a fresh database and for the v1 upgrade alike, so create each store only
|
||||
// when it is missing.
|
||||
@@ -118,11 +122,69 @@ export async function putArtifact(artifact: PersistedArtifact): Promise<void> {
|
||||
// A rejected write (most likely QuotaExceededError) leaves the artifact usable for the
|
||||
// session but unpersisted — degrade like the reads rather than throwing at the caller.
|
||||
await db.put('items', artifact)
|
||||
markSessionDirty(artifact.sessionId, undefined, emailOfScopedKey(ARTIFACTS_DB, db.name))
|
||||
} catch (err) {
|
||||
console.error('Could not persist artifact', err)
|
||||
}
|
||||
}
|
||||
|
||||
/** A session's artifacts with their history, or undefined when the store is unavailable
|
||||
* or no longer the named user's (see `readStoredSessions`). */
|
||||
export async function readSessionArtifacts(
|
||||
sessionId: string,
|
||||
email: string
|
||||
): Promise<{ items: PersistedArtifact[]; versions: ArtifactVersion[] } | undefined> {
|
||||
const db = await getDB()
|
||||
if (!db || db.name !== scopedKeyFor(ARTIFACTS_DB, email)) return undefined
|
||||
try {
|
||||
const items = await db.getAllFromIndex('items', 'by-session', sessionId)
|
||||
const versions = (
|
||||
await Promise.all(items.map((i) => db.getAllFromIndex('versions', 'by-artifact', i.id)))
|
||||
).flat()
|
||||
return { items, versions }
|
||||
} catch (err) {
|
||||
console.error('Could not read artifacts', err)
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
/** Write restored artifacts and snapshots, leaving any that already exist alone. Reports
|
||||
* whether they are all in the store now: unlike the other writes here, a caller records the
|
||||
* restore as done on the strength of this answer. */
|
||||
export async function importArtifacts(
|
||||
items: PersistedArtifact[],
|
||||
versions: ArtifactVersion[],
|
||||
email: string,
|
||||
overwrite = false
|
||||
): Promise<boolean> {
|
||||
if (items.length === 0 && versions.length === 0) return true
|
||||
const db = await getDB()
|
||||
if (!db || db.name !== scopedKeyFor(ARTIFACTS_DB, email)) return false
|
||||
try {
|
||||
const tx = db.transaction(['items', 'versions'], 'readwrite')
|
||||
const itemStore = tx.objectStore('items')
|
||||
const versionStore = tx.objectStore('versions')
|
||||
// An overwrite never puts an older record over a newer one: without a cross-tab lock,
|
||||
// another restore may have landed a newer backup's copy meanwhile.
|
||||
for (const item of items) {
|
||||
const existing = await itemStore.get(item.id)
|
||||
if (existing === undefined || (overwrite && existing.updatedAt <= item.updatedAt)) {
|
||||
await itemStore.put(item)
|
||||
}
|
||||
}
|
||||
for (const version of versions) {
|
||||
if (overwrite || (await versionStore.getKey(version.key)) === undefined) {
|
||||
await versionStore.put(version)
|
||||
}
|
||||
}
|
||||
await tx.done
|
||||
return true
|
||||
} catch (err) {
|
||||
console.error('Could not import artifacts', err)
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
export async function getArtifact(id: string): Promise<PersistedArtifact | undefined> {
|
||||
const db = await getDB()
|
||||
if (!db) return undefined
|
||||
@@ -273,7 +335,11 @@ export async function mutateArtifact(
|
||||
reportFailure = false
|
||||
abort()
|
||||
}
|
||||
return { outcome: await settled, artifact: edit.artifact }
|
||||
const outcome = await settled
|
||||
if (outcome === 'saved' && db) {
|
||||
markSessionDirty(edit.artifact.sessionId, undefined, emailOfScopedKey(ARTIFACTS_DB, db.name))
|
||||
}
|
||||
return { outcome, artifact: edit.artifact }
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -342,14 +408,47 @@ export async function deleteArtifact(id: string): Promise<void> {
|
||||
if (!db) return
|
||||
try {
|
||||
const tx = db.transaction(['items', 'versions'], 'readwrite')
|
||||
await tx.objectStore('items').delete(id)
|
||||
const items = tx.objectStore('items')
|
||||
const sessionId = (await items.get(id))?.sessionId
|
||||
await items.delete(id)
|
||||
await deleteVersionsIn(tx.objectStore('versions'), id)
|
||||
await tx.done
|
||||
if (sessionId) markSessionDirty(sessionId, undefined, emailOfScopedKey(ARTIFACTS_DB, db.name))
|
||||
} catch (err) {
|
||||
console.error('Could not delete artifact', err)
|
||||
}
|
||||
}
|
||||
|
||||
/** Deletes these artifacts of the session (with their versions) and these versions: what
|
||||
* an earlier restore staged for it and the backup no longer has. False when nothing could
|
||||
* be deleted. */
|
||||
export async function pruneSessionArtifacts(
|
||||
sessionId: string,
|
||||
itemIds: Set<string>,
|
||||
versionKeys: Set<string>,
|
||||
email: string
|
||||
): Promise<boolean> {
|
||||
if (itemIds.size === 0 && versionKeys.size === 0) return true
|
||||
const db = await getDB()
|
||||
if (!db || db.name !== scopedKeyFor(ARTIFACTS_DB, email)) return false
|
||||
try {
|
||||
const tx = db.transaction(['items', 'versions'], 'readwrite')
|
||||
const items = tx.objectStore('items')
|
||||
const versions = tx.objectStore('versions')
|
||||
for (const id of await items.index('by-session').getAllKeys(sessionId)) {
|
||||
if (!itemIds.has(String(id))) continue
|
||||
await items.delete(id)
|
||||
await deleteVersionsIn(versions, String(id))
|
||||
}
|
||||
for (const key of versionKeys) await versions.delete(key)
|
||||
await tx.done
|
||||
return true
|
||||
} catch (err) {
|
||||
console.error('Could not prune artifacts for session', err)
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
export async function deleteArtifactsForSession(sessionId: string): Promise<void> {
|
||||
const db = await getDB()
|
||||
if (!db) return
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,204 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
artifactsFingerprint,
|
||||
headSig,
|
||||
jsonBytes,
|
||||
planSessionPush,
|
||||
splitEntry,
|
||||
type ChatSnapshot,
|
||||
type MirrorSyncState
|
||||
} from './sessionMirrorPlan'
|
||||
import type { Session } from './sessionState.svelte'
|
||||
|
||||
function session(over: Partial<Session> = {}): Session {
|
||||
return { id: 's1', name: 'session-1', createdAt: 1, workspace_id: 'ws', chatId: 'c1', ...over }
|
||||
}
|
||||
|
||||
function chat(id: string, lastModified: number, imageIds: string[] = []): ChatSnapshot {
|
||||
return { id, lastModified, record: { id, lastModified }, imageIds }
|
||||
}
|
||||
|
||||
const noArtifacts = { items: [], versions: [] }
|
||||
|
||||
function synced(over: Partial<MirrorSyncState> = {}): MirrorSyncState {
|
||||
return {
|
||||
id: 's1',
|
||||
ws: 'ws',
|
||||
head: headSig(session()),
|
||||
chats: { c1: 10 },
|
||||
images: { i1: 'c1' },
|
||||
artifacts: artifactsFingerprint(noArtifacts),
|
||||
...over
|
||||
}
|
||||
}
|
||||
|
||||
describe('planSessionPush', () => {
|
||||
it('pushes everything for a session never backed up, and nothing for an unsent draft', () => {
|
||||
const plan = planSessionPush({
|
||||
session: session(),
|
||||
chats: [chat('c1', 10, ['i1'])],
|
||||
artifacts: noArtifacts
|
||||
})
|
||||
expect(plan?.entry?.head?.id).toBe('s1')
|
||||
expect(plan?.entry?.chats?.map((c) => c.id)).toEqual(['c1'])
|
||||
expect(plan?.images).toEqual([{ chat_id: 'c1', id: 'i1' }])
|
||||
// Nothing to store yet, so no artifacts object either.
|
||||
expect(plan?.entry?.artifacts).toBeUndefined()
|
||||
expect(plan?.next).toEqual(synced())
|
||||
|
||||
expect(
|
||||
planSessionPush({
|
||||
session: session({ workspace_id: undefined, pending_workspace_id: 'ws' }),
|
||||
chats: [],
|
||||
artifacts: noArtifacts
|
||||
})
|
||||
).toBeUndefined()
|
||||
})
|
||||
|
||||
it('sends nothing when only the fields reading a session bumps changed', () => {
|
||||
const plan = planSessionPush({
|
||||
session: session({ lastSeenCount: 7, lastActivityAt: 99, name: 'session-9' }),
|
||||
chats: [chat('c1', 10, ['i1'])],
|
||||
artifacts: noArtifacts,
|
||||
sync: synced()
|
||||
})
|
||||
expect(plan?.entry).toBeUndefined()
|
||||
expect(plan?.images).toEqual([])
|
||||
})
|
||||
|
||||
it('carries only the chat whose lastModified moved, plus its new images', () => {
|
||||
const plan = planSessionPush({
|
||||
session: session(),
|
||||
chats: [chat('c1', 10, ['i1']), chat('c2', 20, ['i2'])],
|
||||
artifacts: noArtifacts,
|
||||
sync: synced()
|
||||
})
|
||||
expect(plan?.entry?.head).toBeUndefined()
|
||||
expect(plan?.entry?.chats?.map((c) => c.id)).toEqual(['c2'])
|
||||
expect(plan?.images).toEqual([{ chat_id: 'c2', id: 'i2' }])
|
||||
expect(plan?.next.chats).toEqual({ c1: 10, c2: 20 })
|
||||
expect(plan?.next.images).toEqual({ i1: 'c1', i2: 'c2' })
|
||||
})
|
||||
|
||||
it('deletes the copy of a chat that grew too large to back up, instead of keeping a stale one', () => {
|
||||
const plan = planSessionPush({
|
||||
session: session(),
|
||||
chats: [{ id: 'c1', lastModified: 11, imageIds: ['i1'], omitted: true }],
|
||||
artifacts: noArtifacts,
|
||||
sync: synced()
|
||||
})
|
||||
expect(plan?.entry?.delete_chats).toEqual(['c1'])
|
||||
expect(plan?.entry?.chats).toBeUndefined()
|
||||
expect(plan?.images).toEqual([])
|
||||
expect(plan?.next.chats).toEqual({})
|
||||
expect(plan?.next.images).toEqual({})
|
||||
})
|
||||
|
||||
it('carries deletes past the per-entry cap over to the next push', () => {
|
||||
const prevChats = Object.fromEntries(Array.from({ length: 1005 }, (_, i) => [`c${i}`, 10]))
|
||||
const plan = planSessionPush({
|
||||
session: session(),
|
||||
chats: [],
|
||||
artifacts: noArtifacts,
|
||||
sync: synced({ chats: prevChats, images: {} })
|
||||
})
|
||||
expect(plan?.entry?.delete_chats).toHaveLength(1000)
|
||||
// Still listed as pushed, so the next plan finds them gone again, and the session
|
||||
// stays marked for that plan.
|
||||
expect(Object.keys(plan?.next.chats ?? {})).toHaveLength(5)
|
||||
expect(plan?.carried).toBe(true)
|
||||
})
|
||||
|
||||
it('deletes a chat that is gone and an image its chat evicted', () => {
|
||||
const plan = planSessionPush({
|
||||
session: session(),
|
||||
chats: [chat('c1', 11, [])],
|
||||
artifacts: noArtifacts,
|
||||
sync: synced({ chats: { c1: 10, c2: 20 }, images: { i1: 'c1', i2: 'c2' } })
|
||||
})
|
||||
expect(plan?.entry?.delete_chats).toEqual(['c2'])
|
||||
// i2 goes with c2 server-side; only c1's evicted image is deleted on its own.
|
||||
expect(plan?.entry?.delete_images).toEqual([{ chat_id: 'c1', id: 'i1' }])
|
||||
})
|
||||
|
||||
it('moves a session as a full push to the new workspace and a removal from the old', () => {
|
||||
const plan = planSessionPush({
|
||||
session: session({ workspace_id: 'ws2' }),
|
||||
chats: [chat('c1', 10, ['i1'])],
|
||||
artifacts: noArtifacts,
|
||||
sync: synced()
|
||||
})
|
||||
expect(plan?.workspaceId).toBe('ws2')
|
||||
expect(plan?.removeFrom).toBe('ws')
|
||||
expect(plan?.entry?.head?.workspace_id).toBe('ws2')
|
||||
expect(plan?.entry?.chats?.map((c) => c.id)).toEqual(['c1'])
|
||||
expect(plan?.images).toEqual([{ chat_id: 'c1', id: 'i1' }])
|
||||
expect(plan?.next.ws).toBe('ws2')
|
||||
})
|
||||
|
||||
it('pushes artifacts when their fingerprint changes, including emptying them', () => {
|
||||
const items = [
|
||||
{
|
||||
id: 'a1',
|
||||
sessionId: 's1',
|
||||
kind: 'md' as const,
|
||||
name: 'notes',
|
||||
content: 'x',
|
||||
createdAt: 1,
|
||||
updatedAt: 2,
|
||||
version: 1
|
||||
}
|
||||
]
|
||||
const withArtifact = planSessionPush({
|
||||
session: session(),
|
||||
chats: [chat('c1', 10, ['i1'])],
|
||||
artifacts: { items, versions: [] },
|
||||
sync: synced()
|
||||
})
|
||||
expect(withArtifact?.entry?.artifacts).toEqual({ items, versions: [] })
|
||||
|
||||
const emptied = planSessionPush({
|
||||
session: session(),
|
||||
chats: [chat('c1', 10, ['i1'])],
|
||||
artifacts: noArtifacts,
|
||||
sync: synced({ artifacts: artifactsFingerprint({ items, versions: [] }) })
|
||||
})
|
||||
expect(emptied?.entry?.artifacts).toEqual(noArtifacts)
|
||||
})
|
||||
})
|
||||
|
||||
describe('jsonBytes', () => {
|
||||
it('counts the bytes the request carries, not UTF-16 code units', () => {
|
||||
expect(jsonBytes('ab')).toBe(4)
|
||||
expect(jsonBytes('日本')).toBe(8)
|
||||
expect(jsonBytes('😀')).toBe(6)
|
||||
})
|
||||
})
|
||||
|
||||
describe('splitEntry', () => {
|
||||
it('splits an oversized entry into chat-only parts, the head riding on the last', () => {
|
||||
const big = (id: string) => ({ id, record: { id, text: 'x'.repeat(150) } })
|
||||
const entry = {
|
||||
id: 's',
|
||||
head: { id: 's' },
|
||||
chats: [big('c1'), big('c2'), big('c3')],
|
||||
delete_chats: ['old']
|
||||
}
|
||||
const parts = splitEntry(entry, 200)
|
||||
expect(parts.map((p) => p.chats?.map((c) => c.id))).toEqual([['c1'], ['c2'], ['c3']])
|
||||
expect(
|
||||
parts.slice(0, -1).every((p) => p.head === undefined && p.delete_chats === undefined)
|
||||
).toBe(true)
|
||||
expect(parts.at(-1)?.head).toEqual({ id: 's' })
|
||||
expect(parts.at(-1)?.delete_chats).toEqual(['old'])
|
||||
// Within the target, or a single chat: nothing to split.
|
||||
expect(splitEntry(entry, 10_000)).toEqual([entry])
|
||||
expect(splitEntry({ id: 's', chats: [big('c1')] }, 10)).toHaveLength(1)
|
||||
// The server's per-entry chat cap splits too, however small the chats.
|
||||
const many = {
|
||||
id: 's',
|
||||
chats: Array.from({ length: 250 }, (_, i) => ({ id: `c${i}`, record: {} }))
|
||||
}
|
||||
expect(splitEntry(many, 1_000_000).map((p) => p.chats?.length)).toEqual([100, 100, 50])
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,309 @@
|
||||
// The pure half of the session backup: given what the local stores hold for one session
|
||||
// and what was last pushed, decide what the next push carries. Every piece is compared
|
||||
// against its own marker (a signature for the record, `lastModified` for a chat, the id
|
||||
// for a write-once image) so a session that only changed locally in ways the backup
|
||||
// does not keep sends nothing.
|
||||
import type { AISessionBackupPush } from '$lib/gen'
|
||||
import { orderedJsonStringify } from '$lib/utils'
|
||||
import type { Session } from './sessionState.svelte'
|
||||
import type { ArtifactVersion, PersistedArtifact } from '../copilot/chat/artifacts/artifactsDB'
|
||||
|
||||
/** What the backup remembers of a session after a successful push. */
|
||||
export interface MirrorSyncState {
|
||||
id: string
|
||||
/** The workspace whose storage holds the backup. */
|
||||
ws: string
|
||||
/** `headSig` of the record pushed. */
|
||||
head: string
|
||||
/** `lastModified` of each chat pushed, by chat id. */
|
||||
chats: Record<string, number>
|
||||
/** The chat each pushed image belongs to, by image id. */
|
||||
images: Record<string, string>
|
||||
artifacts?: string
|
||||
/** The workspace's storage went away after this push: what it holds is unknown, so the
|
||||
* next push carries everything again. Kept rather than deleted, so a removal still
|
||||
* knows a backup existed. */
|
||||
stale?: boolean
|
||||
/** The dirty mark's counter this push covered. A mark is pending while its counter is
|
||||
* above this; retiring it here rather than deleting the mark means a tab bumping the
|
||||
* counter while another flushes can never have its bump erased. */
|
||||
flushedV?: number
|
||||
/** The user deleted the session and its removal mark could not be written to
|
||||
* localStorage (full): the row itself carries the removal, until it lands. */
|
||||
removed?: boolean
|
||||
/** The storage the push landed in, as the server names it, and the backup generation
|
||||
* (bumped by a workspace key rotation) it landed under. A row recorded against another
|
||||
* storage or generation describes objects the server no longer looks at. */
|
||||
storageId?: string
|
||||
generation?: number
|
||||
/** Other storages this workspace was on that still hold a copy of the backup (a switch
|
||||
* leaves the old copy where it was): a removal is done only once each has answered it,
|
||||
* or a switch back would bring a deleted session back. */
|
||||
alsoIn?: string[]
|
||||
/** Bumps of the dirty mark that localStorage refused, recorded here instead: the mark's
|
||||
* counter plus this is what a push retires, and every row write keeps it. */
|
||||
extraV?: number
|
||||
/** A restore in progress (or cut short): the pieces it wrote for a session that has no
|
||||
* record yet, so a later restore deletes the ones the backup no longer has. */
|
||||
staging?: { chats: string[]; images: string[]; items: string[]; versions: string[] }
|
||||
}
|
||||
|
||||
/**
|
||||
* The part of a session record the backup keeps. Left out on purpose: `name` (a
|
||||
* per-browser counter), the unsent-draft fields (`pending_*`, `draftPrompt`,
|
||||
* `autoSendDraftAt`), `workspace_root_id` (derived on import), `transient`, and the two
|
||||
* fields reading a session bumps (`lastSeenCount`, `lastActivityAt`) — so opening a
|
||||
* session and reading its new messages never costs a push.
|
||||
*/
|
||||
export type SessionHead = Pick<
|
||||
Session,
|
||||
| 'id'
|
||||
| 'workspace_id'
|
||||
| 'chatId'
|
||||
| 'summary'
|
||||
| 'summarySource'
|
||||
| 'createdAt'
|
||||
| 'archived'
|
||||
| 'archivedByWorkspace'
|
||||
| 'moves'
|
||||
| 'previewTabs'
|
||||
| 'activePreviewTabId'
|
||||
| 'previewCollapsed'
|
||||
| 'previewSize'
|
||||
>
|
||||
|
||||
export function sessionHead(s: Session): SessionHead {
|
||||
const head: SessionHead = { id: s.id, createdAt: s.createdAt }
|
||||
if (s.workspace_id !== undefined) head.workspace_id = s.workspace_id
|
||||
if (s.chatId !== undefined) head.chatId = s.chatId
|
||||
if (s.summary !== undefined) head.summary = s.summary
|
||||
if (s.summarySource !== undefined) head.summarySource = s.summarySource
|
||||
if (s.archived !== undefined) head.archived = s.archived
|
||||
if (s.archivedByWorkspace !== undefined) head.archivedByWorkspace = s.archivedByWorkspace
|
||||
if (s.moves !== undefined) head.moves = s.moves
|
||||
if (s.previewTabs !== undefined) head.previewTabs = s.previewTabs
|
||||
if (s.activePreviewTabId !== undefined) head.activePreviewTabId = s.activePreviewTabId
|
||||
if (s.previewCollapsed !== undefined) head.previewCollapsed = s.previewCollapsed
|
||||
if (s.previewSize !== undefined) head.previewSize = s.previewSize
|
||||
return head
|
||||
}
|
||||
|
||||
export function headSig(s: Session): string {
|
||||
return orderedJsonStringify(sessionHead(s))
|
||||
}
|
||||
|
||||
export interface ArtifactsSnapshot {
|
||||
items: PersistedArtifact[]
|
||||
versions: ArtifactVersion[]
|
||||
}
|
||||
|
||||
/** Cheap to compute from the rows alone: every edit bumps `updatedAt`, every snapshot has
|
||||
* its own key, and approving a plan changes `approvedVersion`. */
|
||||
export function artifactsFingerprint(a: ArtifactsSnapshot): string {
|
||||
const items = a.items
|
||||
.map((i) => `${i.id}:${i.updatedAt}:${i.version ?? 1}:${i.approvedVersion ?? ''}`)
|
||||
.sort()
|
||||
const versions = a.versions.map((v) => v.key).sort()
|
||||
return JSON.stringify([items, versions])
|
||||
}
|
||||
|
||||
export interface ChatSnapshot {
|
||||
id: string
|
||||
lastModified: number
|
||||
/** The stored record; absent for a chat that did not change since the last push, whose
|
||||
* bytes the caller did not read. */
|
||||
record?: unknown
|
||||
imageIds: string[]
|
||||
/** Too large to back up: planned as if it did not exist, so a copy pushed while it was
|
||||
* smaller is deleted rather than restored one day as the current transcript. */
|
||||
omitted?: boolean
|
||||
}
|
||||
|
||||
export interface PlanInput {
|
||||
session: Session
|
||||
/** Every chat the session owns right now. */
|
||||
chats: ChatSnapshot[]
|
||||
artifacts: ArtifactsSnapshot
|
||||
sync?: MirrorSyncState
|
||||
}
|
||||
|
||||
export interface PlannedPush {
|
||||
workspaceId: string
|
||||
/** Absent when nothing changed that the backup keeps. */
|
||||
entry?: AISessionBackupPush
|
||||
/** Images the entry needs uploaded, whose bytes the caller loads. */
|
||||
images: { chat_id: string; id: string }[]
|
||||
/** The workspace the session was backed up in before it moved. */
|
||||
removeFrom?: string
|
||||
next: MirrorSyncState
|
||||
/** Deletes past the per-entry cap were left in `next` for the following push, so the
|
||||
* session must stay marked once this one lands. */
|
||||
carried: boolean
|
||||
/** Nothing of the session is taken to be in the storage: every piece goes, and the
|
||||
* first part opens the push whole (see `whole` on the entry). */
|
||||
whole: boolean
|
||||
}
|
||||
|
||||
/** `undefined` for a session with nowhere to go: an unsent draft has no workspace yet. */
|
||||
export function planSessionPush(input: PlanInput): PlannedPush | undefined {
|
||||
const { session, chats, artifacts } = input
|
||||
const workspaceId = session.workspace_id
|
||||
if (!workspaceId) return undefined
|
||||
// A move is a full push into the new workspace's storage; the copy in the old one goes.
|
||||
const prev = input.sync?.ws === workspaceId ? input.sync : undefined
|
||||
const removeFrom = input.sync && input.sync.ws !== workspaceId ? input.sync.ws : undefined
|
||||
|
||||
const entry: AISessionBackupPush = { id: session.id }
|
||||
let changed = false
|
||||
let carried = false
|
||||
const sig = headSig(session)
|
||||
if (prev?.head !== sig) {
|
||||
entry.head = sessionHead(session)
|
||||
changed = true
|
||||
}
|
||||
|
||||
const next: MirrorSyncState = {
|
||||
id: session.id,
|
||||
ws: workspaceId,
|
||||
head: sig,
|
||||
chats: {},
|
||||
images: {}
|
||||
}
|
||||
const images: { chat_id: string; id: string }[] = []
|
||||
const pushedChats: { id: string; record: Record<string, unknown> }[] = []
|
||||
for (const chat of chats) {
|
||||
if (chat.omitted) continue
|
||||
next.chats[chat.id] = chat.lastModified
|
||||
if (prev?.chats[chat.id] !== chat.lastModified && chat.record !== undefined) {
|
||||
pushedChats.push({ id: chat.id, record: chat.record as Record<string, unknown> })
|
||||
}
|
||||
for (const id of chat.imageIds) {
|
||||
next.images[id] = chat.id
|
||||
if (prev?.images[id] === undefined) images.push({ chat_id: chat.id, id })
|
||||
}
|
||||
}
|
||||
if (pushedChats.length > 0) {
|
||||
entry.chats = pushedChats
|
||||
changed = true
|
||||
}
|
||||
if (prev) {
|
||||
const gone = Object.keys(prev.chats).filter((id) => next.chats[id] === undefined)
|
||||
// An image evicted by the per-chat cap, from a chat that is still there (a deleted
|
||||
// chat takes its images with it server-side).
|
||||
const evicted = Object.entries(prev.images).filter(
|
||||
([id, chatId]) => next.images[id] === undefined && next.chats[chatId] !== undefined
|
||||
)
|
||||
// Past the server's cap per entry, the rest stays in `next` as if still pushed, so
|
||||
// the following push finds it gone again.
|
||||
if (gone.length > 0) {
|
||||
entry.delete_chats = gone.slice(0, MAX_DELETES_PER_ENTRY)
|
||||
for (const id of gone.slice(MAX_DELETES_PER_ENTRY)) {
|
||||
next.chats[id] = prev.chats[id]
|
||||
carried = true
|
||||
}
|
||||
changed = true
|
||||
}
|
||||
if (evicted.length > 0) {
|
||||
entry.delete_images = evicted
|
||||
.slice(0, MAX_DELETES_PER_ENTRY)
|
||||
.map(([id, chatId]) => ({ chat_id: chatId, id }))
|
||||
for (const [id, chatId] of evicted.slice(MAX_DELETES_PER_ENTRY)) {
|
||||
next.images[id] = chatId
|
||||
carried = true
|
||||
}
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
|
||||
const fingerprint = artifactsFingerprint(artifacts)
|
||||
next.artifacts = fingerprint
|
||||
if (prev?.artifacts !== fingerprint && (artifacts.items.length > 0 || prev?.artifacts)) {
|
||||
entry.artifacts = { items: artifacts.items, versions: artifacts.versions }
|
||||
changed = true
|
||||
}
|
||||
|
||||
return {
|
||||
workspaceId,
|
||||
entry: changed ? entry : undefined,
|
||||
images,
|
||||
removeFrom,
|
||||
next,
|
||||
carried,
|
||||
whole: prev === undefined
|
||||
}
|
||||
}
|
||||
|
||||
/** Bytes a JSON body would carry for this value, as sent: UTF-8, not UTF-16 code units,
|
||||
* which would under-count a transcript in a non-Latin script by up to three times. Counted
|
||||
* rather than encoded: the values measured are the multi-megabyte ones. */
|
||||
export function jsonBytes(value: unknown): number {
|
||||
const text = JSON.stringify(value)
|
||||
let bytes = 0
|
||||
for (let i = 0; i < text.length; i++) {
|
||||
const c = text.charCodeAt(i)
|
||||
if (c < 0x80) bytes += 1
|
||||
else if (c < 0x800) bytes += 2
|
||||
else if (c >= 0xd800 && c <= 0xdbff) {
|
||||
// A surrogate pair is one four-byte code point.
|
||||
bytes += 4
|
||||
i++
|
||||
} else bytes += 3
|
||||
}
|
||||
return bytes
|
||||
}
|
||||
|
||||
/** Object-store calls the server makes for an entry, the unit its per-request cap counts. */
|
||||
export function operationsOf(entry: AISessionBackupPush): number {
|
||||
return (
|
||||
(entry.chats?.length ?? 0) +
|
||||
(entry.images?.length ?? 0) +
|
||||
(entry.delete_chats?.length ?? 0) +
|
||||
(entry.delete_images?.length ?? 0)
|
||||
)
|
||||
}
|
||||
|
||||
export interface PushBody {
|
||||
owner: string
|
||||
sessions: AISessionBackupPush[]
|
||||
removed?: string[]
|
||||
}
|
||||
|
||||
/** The server's caps on chats and on each delete list per entry. */
|
||||
export const MAX_CHATS_PER_ENTRY = 100
|
||||
export const MAX_DELETES_PER_ENTRY = 1000
|
||||
|
||||
/**
|
||||
* Break an entry that outgrows the target, or the server's per-entry chat cap, into
|
||||
* chat-only entries, each written on its own, with everything else riding on the last one:
|
||||
* the entries go out in order and the server lists the session by the last, so the marker
|
||||
* never lists a chat that has not landed. (A push of the session whole moves the head to
|
||||
* whichever part goes first; see the mirror.)
|
||||
*/
|
||||
export function splitEntry(entry: AISessionBackupPush, targetBytes: number): AISessionBackupPush[] {
|
||||
if (
|
||||
!entry.chats ||
|
||||
entry.chats.length <= 1 ||
|
||||
(entry.chats.length <= MAX_CHATS_PER_ENTRY && jsonBytes(entry) <= targetBytes)
|
||||
) {
|
||||
return [entry]
|
||||
}
|
||||
const { chats, ...rest } = entry
|
||||
const parts: AISessionBackupPush[] = []
|
||||
let current: typeof chats = []
|
||||
let size = 0
|
||||
for (const chat of chats) {
|
||||
const bytes = jsonBytes(chat)
|
||||
if (
|
||||
current.length > 0 &&
|
||||
(current.length >= MAX_CHATS_PER_ENTRY || size + bytes > targetBytes)
|
||||
) {
|
||||
parts.push({ id: entry.id, chats: current })
|
||||
current = []
|
||||
size = 0
|
||||
}
|
||||
current.push(chat)
|
||||
size += bytes
|
||||
}
|
||||
parts.push({ ...rest, chats: current })
|
||||
return parts
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
// The one-way channel from the IndexedDB write funnels (session records, chat history,
|
||||
// artifacts) to the session backup. Import-free on purpose: the stores it is called from
|
||||
// must not depend on the backup module, which depends on all of them.
|
||||
|
||||
// `email` names the user whose store the write landed in (from the store's scoped name):
|
||||
// the current user may have changed while the write was pending, and the mark belongs to
|
||||
// the store's user, not to whoever is logged in when it completes.
|
||||
export type MirrorSignal =
|
||||
| { kind: 'dirty'; sessionId: string; chatId?: string; email?: string }
|
||||
| { kind: 'removed'; sessionId: string; workspaceId?: string; email?: string }
|
||||
|
||||
let handler: ((signal: MirrorSignal) => void) | undefined
|
||||
// Signals raised before the backup module registered, replayed to it on registration.
|
||||
let buffered: MirrorSignal[] = []
|
||||
|
||||
function emit(signal: MirrorSignal): void {
|
||||
if (handler) handler(signal)
|
||||
else buffered.push(signal)
|
||||
}
|
||||
|
||||
/** A durable local write landed for this session (and, when known, this chat) in the
|
||||
* store of `email`. */
|
||||
export function markSessionDirty(sessionId: string, chatId?: string, email?: string): void {
|
||||
emit({ kind: 'dirty', sessionId, chatId, email })
|
||||
}
|
||||
|
||||
/** The user deleted this session; its backup goes with it. */
|
||||
export function markSessionRemoved(sessionId: string, workspaceId?: string, email?: string): void {
|
||||
emit({ kind: 'removed', sessionId, workspaceId, email })
|
||||
}
|
||||
|
||||
export function onMirrorSignal(fn: (signal: MirrorSignal) => void): void {
|
||||
handler = fn
|
||||
const replay = buffered
|
||||
buffered = []
|
||||
for (const signal of replay) fn(signal)
|
||||
}
|
||||
|
||||
export function __resetMirrorSignalForTesting(): void {
|
||||
handler = undefined
|
||||
buffered = []
|
||||
}
|
||||
@@ -24,8 +24,10 @@ import { workspaceRootId } from './sessionScope.svelte'
|
||||
import { clearSessionRecovered } from './sessionRecoveryNotice.svelte'
|
||||
import { type DBSchema, type IDBPDatabase } from 'idb'
|
||||
import { userScopedDb } from '$lib/userScopedDb'
|
||||
import { emailOfScopedKey, scopedKeyFor } from '$lib/userScopedStorage'
|
||||
import { deleteItemsForSession } from '../copilot/chat/files/attachedFilesDB'
|
||||
import { deleteArtifactsForSession } from '../copilot/chat/artifacts/artifactsDB'
|
||||
import { markSessionDirty, markSessionRemoved } from './sessionMirrorSignal'
|
||||
|
||||
// Switch the global workspace iff the target differs from the active one
|
||||
// and is non-empty. Centralises the "session needs its workspace in focus"
|
||||
@@ -102,6 +104,10 @@ export type Session = {
|
||||
// archived (not by the user). Lets reconciliation auto-unarchive the session
|
||||
// when the workspace is unarchived, while leaving user-archived sessions be.
|
||||
archivedByWorkspace?: boolean
|
||||
// How many times the session moved to another workspace. The backup keeps it
|
||||
// with the session's marker, so a restore that finds a copy in two workspaces
|
||||
// (moved, the old copy not yet removed) takes the later one without a clock.
|
||||
moves?: number
|
||||
// In-memory-only flag: the session exists but hasn't been written to
|
||||
// IndexedDB yet. Set at creation, cleared on the first genuine user touch
|
||||
// (typed prompt, workspace/fork pick, preview tab, rename) which persists
|
||||
@@ -444,6 +450,7 @@ async function deleteSessionRow(db: IDBPDatabase<SessionSchema>, id: string): Pr
|
||||
async function putSessionRow(db: IDBPDatabase<SessionSchema>, s: Session): Promise<void> {
|
||||
if (deletedSessionIds.has(s.id)) return
|
||||
await db.put('sessions', s)
|
||||
markSessionDirty(s.id, undefined, emailOfScopedKey(SESSIONS_DB, db.name))
|
||||
}
|
||||
|
||||
// Write-behind a single session record. Transient sessions are in-memory only
|
||||
@@ -775,6 +782,15 @@ export function findEmptyLandingSession(): Session | undefined {
|
||||
)
|
||||
}
|
||||
|
||||
// Session names are a per-browser counter (`session-N`) that the sessions page puts in
|
||||
// its URL, so a new or restored record takes the number after the highest in use.
|
||||
function nextSessionNumber(sessions: Session[]): number {
|
||||
const numbers = sessions
|
||||
.map((s) => /^session-(\d+)$/.exec(s.name)?.[1])
|
||||
.map((n) => (n ? parseInt(n, 10) : 0))
|
||||
return (numbers.length ? Math.max(...numbers) : 0) + 1
|
||||
}
|
||||
|
||||
export function createSession(): Session {
|
||||
// Reuse an existing untouched draft from the active family rather than pile a
|
||||
// blank entry on every `+`, so several pending sessions can still be built up
|
||||
@@ -795,10 +811,7 @@ export function createSession(): Session {
|
||||
return reusable
|
||||
}
|
||||
sessionState.sessions = sessionState.sessions.filter((s) => !isDiscardableDraft(s))
|
||||
const existingNumbers = sessionState.sessions
|
||||
.map((s) => /^session-(\d+)$/.exec(s.name)?.[1])
|
||||
.map((n) => (n ? parseInt(n, 10) : 0))
|
||||
const next = (existingNumbers.length ? Math.max(...existingNumbers) : 0) + 1
|
||||
const next = nextSessionNumber(sessionState.sessions)
|
||||
// Start in the workspace you're in. The one exception: a root you can't
|
||||
// deploy to (locked, no bypass) steers to its dev, since a session there
|
||||
// couldn't edit anything. The picker lets you switch.
|
||||
@@ -1069,6 +1082,7 @@ export async function moveSessionToWorkspace(id: string, newWorkspaceId: string)
|
||||
const s = sessionState.sessions.find((x) => x.id === id)
|
||||
if (!s) return
|
||||
if (s.workspace_id === newWorkspaceId) return
|
||||
if (s.workspace_id !== undefined) s.moves = (s.moves ?? 0) + 1
|
||||
s.workspace_id = newWorkspaceId
|
||||
delete s.pending_workspace_id
|
||||
delete s.pending_fork
|
||||
@@ -1144,9 +1158,63 @@ export function deleteSession(id: string) {
|
||||
// GC any linked files and artifacts persisted for this session.
|
||||
void deleteItemsForSession(id)
|
||||
void deleteArtifactsForSession(id)
|
||||
// Only a delete the user asked for takes the backup with it: the workspace-lifecycle
|
||||
// removals above keep theirs, so a session dropped by a wrong reconcile can be restored.
|
||||
markSessionRemoved(id, s.workspace_id)
|
||||
logFeatureUsage('ai_session', 'deleted', { entityId: id, workspace: s.workspace_id })
|
||||
}
|
||||
|
||||
// --- Session backup support (sessionMirror) ---
|
||||
|
||||
export function isSessionTombstoned(id: string): boolean {
|
||||
return deletedSessionIds.has(id)
|
||||
}
|
||||
|
||||
// Every stored record of the named user, or undefined when the store is unavailable or
|
||||
// already serves someone else: the backup captures its user up front and must not follow
|
||||
// an in-place account switch.
|
||||
export async function readStoredSessions(email: string): Promise<Session[] | undefined> {
|
||||
if (!BROWSER) return undefined
|
||||
const db = await sessionsDb.whenReady()
|
||||
if (!db || db.name !== scopedKeyFor(SESSIONS_DB, email)) return undefined
|
||||
try {
|
||||
return await db.getAll('sessions')
|
||||
} catch (e) {
|
||||
console.error('Failed to read sessions from IndexedDB', e)
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
// Add restored records for sessions this browser does not have, and re-hydrate the list.
|
||||
// A record that exists, or was deleted here, is left alone: the local copy is the newer
|
||||
// one. Returns the ids written.
|
||||
export async function importSessions(records: Session[], email: string): Promise<string[]> {
|
||||
if (!BROWSER) return []
|
||||
const db = await sessionsDb.whenReady()
|
||||
if (!db || db.name !== scopedKeyFor(SESSIONS_DB, email)) return []
|
||||
const imported: string[] = []
|
||||
try {
|
||||
const tx = db.transaction('sessions', 'readwrite')
|
||||
const existing = new Set((await tx.store.getAllKeys()).map(String))
|
||||
let next = nextSessionNumber([...(await tx.store.getAll()), ...sessionState.sessions])
|
||||
for (const r of records) {
|
||||
if (existing.has(r.id) || deletedSessionIds.has(r.id)) continue
|
||||
const record: Session = { ...r, name: `session-${next++}` }
|
||||
delete record.transient
|
||||
delete record.workspace_root_id
|
||||
ensureSessionRootId(record)
|
||||
await tx.store.put(record)
|
||||
imported.push(record.id)
|
||||
}
|
||||
await tx.done
|
||||
} catch (e) {
|
||||
console.error('Failed to import sessions', e)
|
||||
return []
|
||||
}
|
||||
if (imported.length > 0) await hydrateSessions()
|
||||
return imported
|
||||
}
|
||||
|
||||
export function setSessionChatId(sessionId: string, chatId: string) {
|
||||
const s = sessionState.sessions.find((x) => x.id === sessionId)
|
||||
if (s && s.chatId !== chatId) {
|
||||
@@ -1172,6 +1240,7 @@ async function patchStoredSessionChatId(s: Session, chatId: string): Promise<voi
|
||||
stored.chatId = chatId
|
||||
await tx.store.put(stored)
|
||||
await tx.done
|
||||
markSessionDirty(s.id, undefined, emailOfScopedKey(SESSIONS_DB, db.name))
|
||||
return
|
||||
}
|
||||
await tx.done
|
||||
|
||||
@@ -55,6 +55,7 @@ import {
|
||||
__resetDeletedSessionIdsForTesting,
|
||||
setSessionArchived,
|
||||
setSessionPreviewSize,
|
||||
importSessions,
|
||||
type Session
|
||||
} from './sessionState.svelte'
|
||||
|
||||
@@ -688,4 +689,37 @@ describe('sessionState IndexedDB persistence', () => {
|
||||
userStore.set(undefined)
|
||||
await vi.waitFor(() => expect(sessionState.sessions).toEqual([]))
|
||||
})
|
||||
|
||||
// A restored backup must never replace what this browser has, come back after the
|
||||
// user deleted it here, or take a name the sessions page already routes by.
|
||||
it('importSessions adds only unknown, undeleted records under fresh names', async () => {
|
||||
const user = freshUser()
|
||||
await login(user)
|
||||
const local = session({ id: 'local', name: 'session-3', createdAt: 1, summary: 'mine' })
|
||||
await putSession(local)
|
||||
sessionState.sessions.push(local)
|
||||
deleteSession('local')
|
||||
await flush()
|
||||
await putSession(session({ id: 'kept', name: 'session-5', createdAt: 2, summary: 'kept' }))
|
||||
|
||||
const imported = await importSessions(
|
||||
[
|
||||
session({ id: 'local', name: 'session-1', createdAt: 1, summary: 'remote copy' }),
|
||||
session({ id: 'kept', name: 'session-1', createdAt: 2, summary: 'remote copy' }),
|
||||
session({ id: 'new', name: 'session-1', createdAt: 3, workspace_id: 'ws' })
|
||||
],
|
||||
user.email
|
||||
)
|
||||
expect(imported).toEqual(['new'])
|
||||
await vi.waitFor(() =>
|
||||
expect(sessionState.sessions.map((s) => [s.id, s.name])).toEqual([
|
||||
['new', 'session-6'],
|
||||
['kept', 'session-5']
|
||||
])
|
||||
)
|
||||
expect(sessionState.sessions.find((s) => s.id === 'kept')?.summary).toBe('kept')
|
||||
|
||||
// The wrong user's name gets nothing written.
|
||||
expect(await importSessions([session({ id: 'other', createdAt: 4 })], 'nobody@x')).toEqual([])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -28,6 +28,7 @@
|
||||
import ModelPricing from './ModelPricing.svelte'
|
||||
import AiUsagePanel from './AiUsagePanel.svelte'
|
||||
import { setCopilotInfo } from '$lib/aiStore'
|
||||
import { backupSettingsChanged } from '$lib/components/sessions/sessionMirror.svelte'
|
||||
import AIPromptsModal from '../settings/AIPromptsModal.svelte'
|
||||
import { Settings } from 'lucide-svelte'
|
||||
import { untrack } from 'svelte'
|
||||
@@ -79,6 +80,7 @@
|
||||
let usingOpenaiClientCredentialsOauth = $state(false)
|
||||
let workspaceOverrideEditorOpened = $state(false)
|
||||
let copilotDisabled = $state(false)
|
||||
let sessionsStorageDisabled = $state(false)
|
||||
|
||||
// --- Initial state for dirty tracking ---
|
||||
let initialAiProviders: Exclude<AIConfig['providers'], undefined> = $state({})
|
||||
@@ -90,6 +92,7 @@
|
||||
let initialModelPricing: Record<string, ModelPriceOverride> = $state({})
|
||||
let initialPrompts: Record<string, string> = $state({})
|
||||
let initialCopilotDisabled = $state(false)
|
||||
let initialSessionsStorageDisabled = $state(false)
|
||||
let lastLoadedConfigKey = $state<string | undefined>(undefined)
|
||||
|
||||
function clone<T>(v: T): T {
|
||||
@@ -118,6 +121,7 @@
|
||||
maxTokensPerModel = clone(config?.max_tokens_per_model ?? {})
|
||||
modelPricing = clone(config?.model_pricing ?? {})
|
||||
copilotDisabled = config?.copilot_disabled === true
|
||||
sessionsStorageDisabled = config?.sessions_storage_disabled === true
|
||||
for (const mode of ['edit', 'fix', 'gen']) {
|
||||
if (!(mode in customPrompts)) {
|
||||
customPrompts[mode] = ''
|
||||
@@ -135,6 +139,7 @@
|
||||
initialModelPricing = clone(modelPricing)
|
||||
initialPrompts = clone(customPrompts)
|
||||
initialCopilotDisabled = copilotDisabled
|
||||
initialSessionsStorageDisabled = sessionsStorageDisabled
|
||||
}
|
||||
|
||||
export function loadFromConfig(config: AIConfig | undefined) {
|
||||
@@ -151,6 +156,7 @@
|
||||
maxTokensPerModel = clone(initialMaxTokensPerModel)
|
||||
modelPricing = clone(initialModelPricing)
|
||||
copilotDisabled = initialCopilotDisabled
|
||||
sessionsStorageDisabled = initialSessionsStorageDisabled
|
||||
}
|
||||
|
||||
$effect(() => {
|
||||
@@ -186,7 +192,8 @@
|
||||
JSON.stringify(customPrompts) !== JSON.stringify(initialCustomPrompts) ||
|
||||
JSON.stringify(maxTokensPerModel) !== JSON.stringify(initialMaxTokensPerModel) ||
|
||||
JSON.stringify(modelPricing) !== JSON.stringify(initialModelPricing) ||
|
||||
copilotDisabled !== initialCopilotDisabled
|
||||
copilotDisabled !== initialCopilotDisabled ||
|
||||
sessionsStorageDisabled !== initialSessionsStorageDisabled
|
||||
)
|
||||
|
||||
$effect(() => {
|
||||
@@ -291,8 +298,9 @@
|
||||
.filter(([_, prompt]) => prompt.trim().length > 0)
|
||||
.reduce((acc, [mode, prompt]) => ({ ...acc, [mode]: prompt }), {})
|
||||
|
||||
// The flag is the one thing a workspace on instance defaults still stores of its own.
|
||||
// The flags are what a workspace on instance defaults still stores of its own.
|
||||
const copilot_disabled = copilotDisabled ? true : undefined
|
||||
const sessions_storage_disabled = sessionsStorageDisabled ? true : undefined
|
||||
return Object.keys(aiProviders ?? {}).length > 0
|
||||
? {
|
||||
providers: aiProviders,
|
||||
@@ -303,9 +311,10 @@
|
||||
max_tokens_per_model:
|
||||
Object.keys(maxTokensPerModel).length > 0 ? maxTokensPerModel : undefined,
|
||||
model_pricing: Object.keys(modelPricing).length > 0 ? modelPricing : undefined,
|
||||
copilot_disabled
|
||||
copilot_disabled,
|
||||
sessions_storage_disabled
|
||||
}
|
||||
: { copilot_disabled }
|
||||
: { copilot_disabled, sessions_storage_disabled }
|
||||
}
|
||||
|
||||
function isSaveDisabled(): boolean {
|
||||
@@ -332,6 +341,7 @@
|
||||
|
||||
async function editCopilotConfig(): Promise<void> {
|
||||
const config = buildConfig()
|
||||
const backupsToggled = sessionsStorageDisabled !== initialSessionsStorageDisabled
|
||||
let settingsState: GetCopilotSettingsStateResponse | undefined
|
||||
|
||||
if (customSave) {
|
||||
@@ -348,6 +358,9 @@
|
||||
instance_ai_summary: response.instance_ai_summary
|
||||
}
|
||||
sendUserToast('AI settings updated')
|
||||
// This page's session backups follow the switch at once, rather than at the
|
||||
// next page load.
|
||||
if (backupsToggled) backupSettingsChanged(effectiveWorkspace)
|
||||
}
|
||||
storeInitialState()
|
||||
// Hand the parent what was persisted: it owns `initialConfig`, and this component is
|
||||
@@ -646,6 +659,18 @@
|
||||
options={{ right: 'Hide AI sessions in this workspace' }}
|
||||
/>
|
||||
</SettingCard>
|
||||
<SettingCard
|
||||
label="AI session backups"
|
||||
description="Browsers back their AI sessions up to this workspace's object storage, encrypted with the workspace key, and restore them on a new device or after clearing site data. Nothing is stored while the workspace has no object storage configured. Turn it off to keep sessions in the browser only, for example to spare the storage quota."
|
||||
>
|
||||
<Toggle
|
||||
checked={sessionsStorageDisabled}
|
||||
on:change={(e) => {
|
||||
sessionsStorageDisabled = e.detail
|
||||
}}
|
||||
options={{ right: 'Do not back AI sessions up to the workspace storage' }}
|
||||
/>
|
||||
</SettingCard>
|
||||
{/if}
|
||||
</div>
|
||||
|
||||
|
||||
@@ -1,4 +1,11 @@
|
||||
import { openDB as idbOpenDB, deleteDB as idbDeleteDB, type DBSchema, type IDBPDatabase } from 'idb'
|
||||
import {
|
||||
openDB as idbOpenDB,
|
||||
deleteDB as idbDeleteDB,
|
||||
type DBSchema,
|
||||
type IDBPDatabase,
|
||||
type IDBPTransaction,
|
||||
type StoreNames
|
||||
} from 'idb'
|
||||
import { scopedKey } from '$lib/userScopedStorage'
|
||||
|
||||
// Per-user IndexedDB lifecycle, shared by the session list and the copilot
|
||||
@@ -21,7 +28,12 @@ export interface UserScopedDbMigrateDeps {
|
||||
|
||||
export interface UserScopedDbOptions<Schema extends DBSchema> {
|
||||
version: number
|
||||
upgrade: (db: IDBPDatabase<Schema>) => void
|
||||
// The version-change transaction is the only way to add an index to a store that
|
||||
// already exists; a store being created gets it from the store handle instead.
|
||||
upgrade: (
|
||||
db: IDBPDatabase<Schema>,
|
||||
tx: IDBPTransaction<Schema, StoreNames<Schema>[], 'versionchange'>
|
||||
) => void
|
||||
// Invoked once per scoped name right after a successful open. The fn owns its
|
||||
// own "already migrated / not applicable" gate (e.g. checking a store's
|
||||
// count) — claim-then-delete legacy data lives here.
|
||||
@@ -100,11 +112,11 @@ export function userScopedDb<Schema extends DBSchema>(
|
||||
try {
|
||||
let handle: IDBPDatabase<Schema> | undefined
|
||||
const db = await openDB<Schema>(name, opts.version, {
|
||||
upgrade(database) {
|
||||
upgrade(database, _oldVersion, _newVersion, transaction) {
|
||||
// The version-change transaction is ours: nothing is queued ahead of this
|
||||
// open any more, and what remains is our own upgrade running.
|
||||
stopWaiting()
|
||||
opts.upgrade(database)
|
||||
opts.upgrade(database, transaction)
|
||||
},
|
||||
// Another tab is opening this database at a higher version, which our open
|
||||
// connection would block indefinitely. Let go so their upgrade lands; this
|
||||
|
||||
@@ -52,7 +52,19 @@ export function getCurrentUserEmail(): string | undefined {
|
||||
// treat that as "do not read/write" so we never touch a browser-global key.
|
||||
export function scopedKey(base: string): string | undefined {
|
||||
if (!currentEmail) return undefined
|
||||
return `${base}::${currentEmail}`
|
||||
return scopedKeyFor(base, currentEmail)
|
||||
}
|
||||
|
||||
// The key a base name has for a given user, for work that captured its user up front and
|
||||
// must not follow an in-place account switch (the session backup flush).
|
||||
export function scopedKeyFor(base: string, email: string): string {
|
||||
return `${base}::${email}`
|
||||
}
|
||||
|
||||
// The email a scoped key or database name was built for, so a write that landed in a
|
||||
// store can name the user it belongs to even after the current user changed.
|
||||
export function emailOfScopedKey(base: string, key: string): string | undefined {
|
||||
return key.startsWith(`${base}::`) ? key.slice(base.length + 2) : undefined
|
||||
}
|
||||
|
||||
// Register a callback invoked whenever the scoping email changes. Fired once
|
||||
|
||||
@@ -92,6 +92,7 @@
|
||||
import { parsePreviewItemRoute } from '$lib/components/sessions/previewPaths'
|
||||
import { rememberNavRoute } from '$lib/components/sessions/sessionSwitch.svelte'
|
||||
import { sessionState } from '$lib/components/sessions/sessionState.svelte'
|
||||
import { restoreSessionBackups } from '$lib/components/sessions/sessionMirror.svelte'
|
||||
import { currentWorkspaceRootId } from '$lib/components/sessions/sessionScope.svelte'
|
||||
import WorkspaceScopeHeader from '$lib/components/sidebar/WorkspaceScopeHeader.svelte'
|
||||
import { DEFAULT_HUB_BASE_URL } from '$lib/hub'
|
||||
@@ -725,6 +726,16 @@
|
||||
$workspaceStore
|
||||
untrack(() => updateUserStore($workspaceStore))
|
||||
})
|
||||
// Bring back the AI sessions this browser lacks for the workspace family in view, once
|
||||
// the local list is known (so nothing it has is fetched again) and the memberships have
|
||||
// resolved (the family is derived from them).
|
||||
$effect(() => {
|
||||
const ws = $workspaceStore
|
||||
const ready = sessionState.hydrated && $usersWorkspaceStore !== undefined
|
||||
if (globalAiEnabled && ready && ws && !$userStore?.operator) {
|
||||
untrack(() => restoreSessionBackups(ws))
|
||||
}
|
||||
})
|
||||
// While a fork is reachable, mirror its parent linkage to localStorage so a
|
||||
// later reload landing on a now-deleted fork can return to the parent (see
|
||||
// forkParentMemory + the deleted-fork recovery in the root layout).
|
||||
|
||||
Reference in New Issue
Block a user