diff --git a/apps/elf-api/src/routes.rs b/apps/elf-api/src/routes.rs index 9ccf7828..0439dc81 100644 --- a/apps/elf-api/src/routes.rs +++ b/apps/elf-api/src/routes.rs @@ -72,7 +72,9 @@ use elf_service::{ SpaceGrantsListRequest, TextPositionSelector, TextQuoteSelector, TraceBundleGetRequest, TraceBundleResponse, TraceGetRequest, TraceGetResponse, TraceRecentListRequest, TraceRecentListResponse, TraceTrajectoryGetRequest, UnpublishNoteRequest, UpdateRequest, - UpdateResponse, search::TraceBundleMode, + UpdateResponse, WorkJournalEntryCreateRequest, WorkJournalEntryCreateResponse, + WorkJournalEntryFamily, WorkJournalEntryGetRequest, WorkJournalEntryResponse, + WorkJournalSessionReadbackRequest, WorkJournalSessionReadbackResponse, search::TraceBundleMode, }; /// JSON OpenAPI contract route. @@ -139,6 +141,9 @@ const VIEWER_HTML: &str = include_str!("../static/viewer.html"); notes_delete, notes_publish, notes_unpublish, + work_journal_entry_create, + work_journal_entry_get, + work_journal_session_readback, space_grants_list, space_grant_upsert, space_grant_revoke, @@ -193,6 +198,7 @@ const VIEWER_HTML: &str = include_str!("../static/viewer.html"); (name = "dreaming", description = "Dreaming review queue and derived memory organization."), (name = "recall", description = "Cross-layer recall and debug readback."), (name = "knowledge", description = "Derived knowledge page rebuild and lint readback."), + (name = "work_journal", description = "Source-adjacent Work Journal capture and session readback."), (name = "admin", description = "Local admin and operator inspection routes."), ) )] @@ -240,6 +246,34 @@ struct DocsPutBody { content: String, } +#[derive(Clone, Debug, Deserialize)] +struct WorkJournalEntryCreateBody { + entry_id: Option, + scope: String, + session_id: String, + family: WorkJournalEntryFamily, + title: Option, + body: String, + source_refs: Vec, + write_policy: Option, + #[serde(default)] + explicit_next_steps: Vec, + #[serde(default)] + inferred_next_steps: Vec, + #[serde(default)] + rejected_options: Vec, + #[serde(default = "empty_json_object")] + promotion_boundary: Value, +} + +#[derive(Clone, Debug, Deserialize)] +struct WorkJournalSessionReadbackBody { + session_id: String, + #[serde(default)] + families: Vec, + limit: Option, +} + #[derive(Clone, Debug, Deserialize)] struct CoreBlockUpsertBody { block_id: Option, @@ -730,6 +764,9 @@ pub fn router(state: AppState) -> Router { ) .route("/v2/notes/{note_id}/publish", routing::post(notes_publish)) .route("/v2/notes/{note_id}/unpublish", routing::post(notes_unpublish)) + .route("/v2/work-journal/entries", routing::post(work_journal_entry_create)) + .route("/v2/work-journal/entries/{entry_id}", routing::get(work_journal_entry_get)) + .route("/v2/work-journal/readback", routing::post(work_journal_session_readback)) .route( "/v2/spaces/{space}/grants", routing::get(space_grants_list).post(space_grant_upsert), @@ -1485,6 +1522,138 @@ async fn docs_put( Ok(Json(response)) } +#[utoipa::path( + post, + path = "/v2/work-journal/entries", + tag = "work_journal", + request_body = Value, + responses( + (status = 200, description = "Work Journal entry was stored.", body = Value), + (status = 400, description = "Invalid request.", body = ErrorBody), + (status = 401, description = "Authentication required.", body = ErrorBody), + (status = 403, description = "Scope denied.", body = ErrorBody), + (status = 422, description = "Non-English input rejected.", body = ErrorBody), + (status = 500, description = "Internal error.", body = ErrorBody), + ) +)] +async fn work_journal_entry_create( + State(state): State, + headers: HeaderMap, + role: Option>, + payload: Result, JsonRejection>, +) -> Result, ApiError> { + let ctx = RequestContext::from_headers(&headers)?; + let Json(payload) = payload.map_err(|err| { + tracing::warn!(error = %err, "Invalid request payload."); + + json_error(StatusCode::BAD_REQUEST, "INVALID_REQUEST", "Invalid request payload.", None) + })?; + let role = role.map(|Extension(role)| role); + + if payload.scope.trim() == "org_shared" { + require_admin_for_org_shared_writes(state.service.cfg.security.auth_mode.as_str(), role)?; + } + + let response = state + .service + .work_journal_entry_create(WorkJournalEntryCreateRequest { + tenant_id: ctx.tenant_id, + project_id: ctx.project_id, + agent_id: ctx.agent_id, + entry_id: payload.entry_id, + scope: payload.scope, + session_id: payload.session_id, + family: payload.family, + title: payload.title, + body: payload.body, + source_refs: payload.source_refs, + write_policy: payload.write_policy, + explicit_next_steps: payload.explicit_next_steps, + inferred_next_steps: payload.inferred_next_steps, + rejected_options: payload.rejected_options, + promotion_boundary: payload.promotion_boundary, + }) + .await?; + + Ok(Json(response)) +} + +#[utoipa::path( + get, + path = "/v2/work-journal/entries/{entry_id}", + tag = "work_journal", + responses( + (status = 200, description = "Work Journal entry metadata.", body = Value), + (status = 400, description = "Invalid request.", body = ErrorBody), + (status = 401, description = "Authentication required.", body = ErrorBody), + (status = 403, description = "Scope denied.", body = ErrorBody), + (status = 404, description = "Work Journal entry not found.", body = ErrorBody), + (status = 500, description = "Internal error.", body = ErrorBody), + ) +)] +async fn work_journal_entry_get( + State(state): State, + headers: HeaderMap, + Path(entry_id): Path, +) -> Result, ApiError> { + let ctx = RequestContext::from_headers(&headers)?; + let read_profile = required_read_profile(&headers)?; + let response = state + .service + .work_journal_entry_get(WorkJournalEntryGetRequest { + tenant_id: ctx.tenant_id, + project_id: ctx.project_id, + agent_id: ctx.agent_id, + read_profile, + entry_id, + }) + .await?; + + Ok(Json(response)) +} + +#[utoipa::path( + post, + path = "/v2/work-journal/readback", + tag = "work_journal", + request_body = Value, + responses( + (status = 200, description = "Work Journal session readback.", body = Value), + (status = 400, description = "Invalid request.", body = ErrorBody), + (status = 401, description = "Authentication required.", body = ErrorBody), + (status = 403, description = "Scope denied.", body = ErrorBody), + (status = 422, description = "Non-English input rejected.", body = ErrorBody), + (status = 500, description = "Internal error.", body = ErrorBody), + ) +)] +async fn work_journal_session_readback( + State(state): State, + headers: HeaderMap, + payload: Result, JsonRejection>, +) -> Result, ApiError> { + let ctx = RequestContext::from_headers(&headers)?; + let read_profile = required_read_profile(&headers)?; + let Json(payload) = payload.map_err(|err| { + tracing::warn!(error = %err, "Invalid request payload."); + + json_error(StatusCode::BAD_REQUEST, "INVALID_REQUEST", "Invalid request payload.", None) + })?; + let response = state + .service + .work_journal_session_readback(WorkJournalSessionReadbackRequest { + tenant_id: ctx.tenant_id, + project_id: ctx.project_id, + agent_id: ctx.agent_id, + read_profile, + session_id: payload.session_id, + families: payload.families, + limit: payload.limit, + }) + .await?; + + Ok(Json(response)) +} + #[utoipa::path( get, path = "/v2/core-blocks", diff --git a/apps/elf-api/tests/http.rs b/apps/elf-api/tests/http.rs index e28f6cb7..b18ccd05 100644 --- a/apps/elf-api/tests/http.rs +++ b/apps/elf-api/tests/http.rs @@ -947,6 +947,9 @@ async fn openapi_json_route_serves_generated_contract() { assert_openapi_method(&spec, "/v2/core-blocks", "get"); assert_openapi_method(&spec, "/v2/entity-memory", "get"); assert_openapi_method(&spec, "/v2/docs/search/l0", "post"); + assert_openapi_method(&spec, "/v2/work-journal/entries", "post"); + assert_openapi_method(&spec, "/v2/work-journal/entries/{entry_id}", "get"); + assert_openapi_method(&spec, "/v2/work-journal/readback", "post"); assert_openapi_method(&spec, "/v2/searches/{search_id}/notes", "post"); assert_openapi_method(&spec, "/v2/admin/core-blocks", "post"); assert_openapi_method(&spec, "/v2/admin/core-blocks/{block_id}/attachments", "post"); @@ -997,6 +1000,7 @@ async fn scalar_docs_route_serves_api_reference_html() { assert!(html.contains("/v2/admin/docs/search/l0")); assert!(html.contains("/v2/admin/knowledge/pages")); assert!(html.contains("/v2/admin/knowledge/pages/search")); + assert!(html.contains("/v2/work-journal/readback")); } #[tokio::test] diff --git a/apps/elf-mcp/src/server.rs b/apps/elf-mcp/src/server.rs index c68c7a98..ba0b9867 100644 --- a/apps/elf-mcp/src/server.rs +++ b/apps/elf-mcp/src/server.rs @@ -343,6 +343,48 @@ impl ElfMcp { self.forward(HttpMethod::Post, "/v2/docs/excerpts", params, None).await } + #[rmcp::tool( + name = "elf_work_journal_entry_create", + description = "Capture one source-adjacent Work Journal entry with source refs, redaction, next-step, rejected-option, and promotion-boundary metadata. Journal content is not authoritative memory.", + input_schema = work_journal_entry_create_schema() + )] + async fn elf_work_journal_entry_create( + &self, + params: JsonObject, + ) -> Result { + self.forward(HttpMethod::Post, "/v2/work-journal/entries", params, None).await + } + + #[rmcp::tool( + name = "elf_work_journal_entry_get", + description = "Fetch one readable Work Journal entry by entry_id.", + input_schema = work_journal_entry_get_schema() + )] + async fn elf_work_journal_entry_get( + &self, + mut params: JsonObject, + ) -> Result { + let entry_id = take_required_string(&mut params, "entry_id")?; + let path = format!("/v2/work-journal/entries/{entry_id}"); + + self.forward(HttpMethod::Get, &path, JsonObject::new(), None).await + } + + #[rmcp::tool( + name = "elf_work_journal_session_readback", + description = "Read newest Work Journal entries for a session and return a where_stopped projection with journal evidence. Current-fact answers still require accepted memory or knowledge authority.", + input_schema = work_journal_session_readback_schema() + )] + async fn elf_work_journal_session_readback( + &self, + mut params: JsonObject, + ) -> Result { + // read_profile is part of the MCP server configuration and is not client-controlled. + let _ = take_optional_string(&mut params, "read_profile")?; + + self.forward(HttpMethod::Post, "/v2/work-journal/readback", params, None).await + } + #[rmcp::tool( name = "elf_core_blocks_get", description = "Fetch core memory blocks explicitly attached to the configured agent and read profile. This is separate from archival search.", @@ -1303,6 +1345,97 @@ fn docs_excerpts_get_schema() -> Arc { })) } +fn work_journal_entry_create_schema() -> Arc { + Arc::new(rmcp::object!({ + "type": "object", + "additionalProperties": true, + "required": ["scope", "session_id", "family", "body", "source_refs"], + "properties": { + "entry_id": { "type": ["string", "null"] }, + "scope": { "type": "string", "enum": ["agent_private", "project_shared", "org_shared"] }, + "session_id": { "type": "string" }, + "family": { + "type": "string", + "enum": [ + "session_log", + "handoff_brief", + "janitor_report", + "explicit_next_step", + "inferred_next_step", + "rejected_option" + ] + }, + "title": { "type": ["string", "null"] }, + "body": { "type": "string" }, + "source_refs": { + "type": "array", + "items": { "type": "object", "additionalProperties": true }, + "minItems": 1 + }, + "write_policy": { "type": ["object", "null"] }, + "explicit_next_steps": { + "type": "array", + "items": { "type": "string" } + }, + "inferred_next_steps": { + "type": "array", + "items": { "type": "string" } + }, + "rejected_options": { + "type": "array", + "items": { "type": "string" } + }, + "promotion_boundary": { + "type": ["object", "null"], + "additionalProperties": true, + "properties": { + "authoritative_memory_allowed": { "type": "boolean" }, + "accepted_memory_authority_ref": { "type": "object", "additionalProperties": true }, + "accepted_dreaming_review_ref": { "type": "object", "additionalProperties": true } + } + } + } + })) +} + +fn work_journal_entry_get_schema() -> Arc { + Arc::new(rmcp::object!({ + "type": "object", + "additionalProperties": true, + "required": ["entry_id"], + "properties": { + "entry_id": { "type": "string" } + } + })) +} + +fn work_journal_session_readback_schema() -> Arc { + Arc::new(rmcp::object!({ + "type": "object", + "additionalProperties": true, + "required": ["session_id"], + "properties": { + "session_id": { "type": "string" }, + "families": { + "type": "array", + "items": { + "type": "string", + "enum": [ + "session_log", + "handoff_brief", + "janitor_report", + "explicit_next_step", + "inferred_next_step", + "rejected_option" + ] + } + }, + "limit": { "type": ["integer", "null"] }, + "read_profile": { "type": ["string", "null"] } + } + })) +} + fn core_blocks_get_schema() -> Arc { Arc::new(rmcp::object!({ "type": "object", @@ -1790,7 +1923,7 @@ mod tests { type RequestRecorder = Arc>>>; - const ALL_TOOL_DEFINITIONS: [ToolDefinition; 34] = [ + const ALL_TOOL_DEFINITIONS: [ToolDefinition; 37] = [ ToolDefinition::new( "elf_notes_ingest", HttpMethod::Post, @@ -1845,6 +1978,24 @@ mod tests { "/v2/recall-debug/panel", "Build an agent-facing cross-layer recall/debug panel and deterministic recall_trace over memory traces, source documents, knowledge pages, graph facts, and Dreaming proposals.", ), + ToolDefinition::new( + "elf_work_journal_entry_create", + HttpMethod::Post, + "/v2/work-journal/entries", + "Capture one source-adjacent Work Journal entry with source refs, redaction, next-step, rejected-option, and promotion-boundary metadata.", + ), + ToolDefinition::new( + "elf_work_journal_entry_get", + HttpMethod::Get, + "/v2/work-journal/entries/{entry_id}", + "Fetch one readable Work Journal entry by entry_id.", + ), + ToolDefinition::new( + "elf_work_journal_session_readback", + HttpMethod::Post, + "/v2/work-journal/readback", + "Read newest Work Journal entries for a session and return a where_stopped projection with journal evidence.", + ), ToolDefinition::new( "elf_searches_get", HttpMethod::Get, @@ -2053,6 +2204,9 @@ mod tests { "elf_admin_traces_recent_list", "elf_dreaming_review_queue", "elf_recall_debug_panel", + "elf_work_journal_entry_create", + "elf_work_journal_entry_get", + "elf_work_journal_session_readback", "elf_admin_trace_get", "elf_admin_trajectory_get", "elf_admin_trace_item_get", @@ -2362,6 +2516,34 @@ mod tests { } } + #[test] + fn work_journal_schemas_include_families_and_source_refs() { + let create_schema = super::work_journal_entry_create_schema(); + let create_properties = create_schema + .get("properties") + .and_then(serde_json::Value::as_object) + .expect("work_journal_entry_create schema is missing properties."); + let readback_schema = super::work_journal_session_readback_schema(); + let readback_properties = readback_schema + .get("properties") + .and_then(serde_json::Value::as_object) + .expect("work_journal_session_readback schema is missing properties."); + + for field in ["scope", "session_id", "family", "body", "source_refs"] { + assert!( + create_schema.get("required").and_then(serde_json::Value::as_array).is_some_and( + |fields| { fields.iter().any(|value| value.as_str() == Some(field)) } + ), + "Missing Work Journal required field {field}." + ); + } + + assert!(create_properties.contains_key("write_policy")); + assert!(create_properties.contains_key("promotion_boundary")); + assert!(readback_properties.contains_key("session_id")); + assert!(readback_properties.contains_key("families")); + } + #[test] fn docs_excerpts_get_schema_includes_l0_level_and_optional_explain() { let schema = super::docs_excerpts_get_schema(); diff --git a/docs/evidence/2026-06-27-work-journal-drift-audit.md b/docs/evidence/2026-06-27-work-journal-drift-audit.md new file mode 100644 index 00000000..56e44ced --- /dev/null +++ b/docs/evidence/2026-06-27-work-journal-drift-audit.md @@ -0,0 +1,118 @@ +--- +type: Drift Audit +title: "Work Journal Drift Audit" +description: "Drift audit for source-adjacent Work Journal capture, readback, and promotion-boundary behavior." +resource: docs/evidence/2026-06-27-work-journal-drift-audit.md +status: active +authority: current_state +owner: docs +last_verified: 2026-06-27 +tags: + - docs + - drift-audit + - work-journal +source_refs: + - https://linear.app/hackink/issue/XY-1117 +code_refs: + - packages/elf-service/src/work_journal.rs + - packages/elf-storage/src/work_journal.rs + - packages/elf-storage/src/models.rs + - sql/tables/042_work_journal_entries.sql + - apps/elf-api/src/routes.rs + - apps/elf-mcp/src/server.rs + - packages/elf-service/tests/acceptance/work_journal.rs +related: + - docs/spec/system_work_journal_v1.md + - docs/spec/system_elf_memory_service_v2.md + - docs/spec/system_version_registry.md +drift_watch: + - packages/elf-service/src/work_journal.rs + - packages/elf-storage/src/work_journal.rs + - packages/elf-storage/src/models.rs + - sql/tables/042_work_journal_entries.sql + - apps/elf-api/src/routes.rs + - apps/elf-mcp/src/server.rs + - docs/spec/system_work_journal_v1.md +--- +# Work Journal Drift Audit + +Purpose: Anchor the Work Journal v1 source-adjacent capture and readback contract to +the current service, storage, HTTP, MCP, and test surfaces. +Read this when: You need evidence behind Work Journal session logs, handoff briefs, +janitor reports, next-step captures, rejected options, source refs, redaction, or +promotion-boundary behavior. +Not this document: Memory Authority note promotion, Dreaming Review proposal scoring, +or benchmark competitor interpretation. + +## Watched Claims + +- `work_journal_entries` is the source-of-truth table for Work Journal rows. +- Work Journal entries use stable UUID `entry_id` values and session grouping through + `session_id`. +- Canonical families are `session_log`, `handoff_brief`, `janitor_report`, + `explicit_next_step`, `inferred_next_step`, and `rejected_option`. +- Work Journal capture requires non-empty object source refs and stores redacted + durable journal output plus redaction audit. +- Work Journal readback exposes one-entry lookup and session-level `where_stopped` + evidence without writing authoritative memory notes, indexing outbox rows, search + sessions, traces, or Qdrant points. +- `authoritative_memory_allowed` is true only for supported accepted Memory Authority + or Dreaming Review references that resolve to current storage evidence; caller- + requested authority without accepted promotion evidence remains source-adjacent only. +- HTTP and MCP expose the Work Journal create, get, and session readback surfaces. + +## Evidence Anchors + +- `sql/tables/042_work_journal_entries.sql` defines the Work Journal table and + storage-level scope, family, status, JSON-shape checks, and session/scope indexes. +- `packages/elf-storage/src/models.rs` and `packages/elf-storage/src/work_journal.rs` + own row shape, insert, lookup, and newest-first session queries. +- `packages/elf-service/src/work_journal.rs` owns validation, write-policy redaction, + accepted-reference normalization and storage resolution, read authorization, and + `where_stopped` shaping. +- `apps/elf-api/src/routes.rs` exposes: + - `POST /v2/work-journal/entries` + - `GET /v2/work-journal/entries/{entry_id}` + - `POST /v2/work-journal/readback` +- `apps/elf-mcp/src/server.rs` exposes: + - `elf_work_journal_entry_create` + - `elf_work_journal_entry_get` + - `elf_work_journal_session_readback` +- `packages/elf-service/tests/acceptance/work_journal.rs` checks persistence, + redacted readback, source refs, `where_stopped`, and no Memory Ledger or indexing + side effects. + +## Reverse Checks + +- Run `cargo make check-docs` after Work Journal docs changes. +- Run the focused Work Journal service tests after service validation or readback + changes. +- Run the registered repository gate before review handoff. +- Re-check `elf.memory_record_ref/v1`, Memory Authority readability, and Dreaming + Review accepted-reference resolution if Memory Authority or Dreaming Review target + refs change. +- Re-check HTTP OpenAPI and MCP tool registration if Work Journal route or tool names + change. + +## Verdict + +pass + +## Required Updates + +- Update `docs/spec/system_work_journal_v1.md`, this drift audit, and + `docs/spec/system_version_registry.md` when Work Journal fields, canonical family + values, accepted-reference shapes or storage resolution rules, readback ordering, + or `where_stopped` semantics change. +- Do not treat Work Journal readback as current-fact authority unless the response is + backed by accepted Memory Authority, graph, Knowledge Workspace, or reviewed + Dreaming evidence. + +## Citations + +- `sql/tables/042_work_journal_entries.sql` +- `packages/elf-storage/src/work_journal.rs` +- `packages/elf-service/src/work_journal.rs` +- `apps/elf-api/src/routes.rs` +- `apps/elf-mcp/src/server.rs` +- `packages/elf-service/tests/acceptance/work_journal.rs` diff --git a/docs/evidence/index.md b/docs/evidence/index.md index d5c3f310..bc0a38ab 100644 --- a/docs/evidence/index.md +++ b/docs/evidence/index.md @@ -25,5 +25,7 @@ Routes to: Drift audits and evidence concepts under `docs/evidence/`. - `2026-06-23-privacy-delete-export-boundaries-drift-audit.md`: Drift audit for privacy, delete/forget, export, retention, source visibility, and graph evidence suppression boundaries. +- `2026-06-27-work-journal-drift-audit.md`: Drift audit for Work Journal + source-adjacent capture, readback, redaction, and promotion-boundary behavior. - `external_memory_pattern_radar_latest.md`: Latest weekly external memory pattern radar summary. diff --git a/docs/log.md b/docs/log.md index d72d1888..c6e1cced 100644 --- a/docs/log.md +++ b/docs/log.md @@ -121,3 +121,17 @@ logs. plus a drift audit and spec updates for Source Library span suppression, Knowledge Workspace source visibility, graph evidence readback, and relation-context evidence filtering. + +## 2026-06-27 + +- Added `docs/spec/system_work_journal_v1.md` for XY-1117, defining + source-adjacent Work Journal capture and readback, canonical entry families, + redaction, source refs, promotion-boundary metadata, and the non-authoritative + memory boundary. +- Linked the Work Journal contract from the spec index, version registry, and ELF v2 + service/MCP endpoint map. +- Added the Work Journal drift audit tying the new service, storage, HTTP, MCP, and + test surfaces to the source-adjacent promotion-boundary contract. +- Tightened the Work Journal promotion-boundary contract so accepted references must + resolve to storage-backed Memory Authority or Dreaming Review evidence instead of + granting authority from JSON shape alone. diff --git a/docs/spec/index.md b/docs/spec/index.md index 695c9c54..7660f68a 100644 --- a/docs/spec/index.md +++ b/docs/spec/index.md @@ -50,6 +50,7 @@ Question this index answers: "what must remain true?" - `system_search_filter_expr_v1.md`: System: Search Filter Expression Contract v1. - `system_source_ref_doc_pointer_v1.md`: System: `source_ref` Doc Pointer Resolver (v1). - `system_version_registry.md`: System Version Registry. +- `system_work_journal_v1.md`: Work Journal v1 Specification. ## Spec document contract Start each spec with a compact routing header: diff --git a/docs/spec/system_elf_memory_service_v2.md b/docs/spec/system_elf_memory_service_v2.md index a2b3abd7..6221b16b 100644 --- a/docs/spec/system_elf_memory_service_v2.md +++ b/docs/spec/system_elf_memory_service_v2.md @@ -1905,6 +1905,23 @@ Behavior: verifies the requested chunk, quote, or position selector against current source content. +Work Journal capture and readback: +- POST /v2/work-journal/entries +- GET /v2/work-journal/entries/{entry_id} +- POST /v2/work-journal/readback + +Behavior: +- These endpoints persist and read source-adjacent Work Journal entries for session + logs, handoff briefs, janitor reports, explicit next steps, inferred next steps, + and rejected options. +- Work Journal rows carry source refs, redaction audit, and promotion-boundary + metadata, but they are not authoritative memory. They must not write + `memory_notes`, `indexing_outbox`, search sessions, traces, or Qdrant points. +- Session readback may answer "where did we stop?" with journal evidence. Current + fact answers must still route through accepted Memory Authority, Knowledge + Workspace, graph, or reviewed Dreaming surfaces. +- The detailed contract is defined in `system_work_journal_v1.md`. + GET /v2/admin/events/ingestion-profiles Headers: @@ -2548,6 +2565,9 @@ Original query: - elf_docs_delete -> DELETE /v2/docs/{doc_id} - elf_docs_search_l0 -> POST /v2/docs/search/l0 - elf_docs_excerpts_get -> POST /v2/docs/excerpts + - elf_work_journal_entry_create -> POST /v2/work-journal/entries + - elf_work_journal_entry_get -> GET /v2/work-journal/entries/{entry_id} + - elf_work_journal_session_readback -> POST /v2/work-journal/readback - elf_notes_list -> GET /v2/notes - elf_notes_get -> GET /v2/notes/{note_id} - elf_notes_patch -> PATCH /v2/notes/{note_id} diff --git a/docs/spec/system_version_registry.md b/docs/spec/system_version_registry.md index 367b2658..4eed26ae 100644 --- a/docs/spec/system_version_registry.md +++ b/docs/spec/system_version_registry.md @@ -197,6 +197,28 @@ This document is normative. When a new versioned identifier is introduced, it mu - Bump rule: Introduce a new identifier only if trace entry states, source-ref fields, policy-reason semantics, or deterministic ordering become incompatible. +### Work Journal readback schema + +- Identifier: `elf.work_journal/v1`. +- Type: Source-adjacent Work Journal entry and session readback envelope. +- Defined in: `packages/elf-service/src/work_journal.rs` + (`ELF_WORK_JOURNAL_SCHEMA_V1`) and `docs/spec/system_work_journal_v1.md`. +- Consumers: HTTP routes under `/v2/work-journal/*`, MCP Work Journal tools, and + agents answering "where did we stop?" with session-log evidence. +- Bump rule: Introduce a new identifier only when entry fields, family semantics, + readback ordering, or where-stopped projection semantics become incompatible. + +### Work Journal promotion boundary schema + +- Identifier: `elf.work_journal.promotion_boundary/v1`. +- Type: Promotion-boundary metadata for source-adjacent journal entries. +- Defined in: `packages/elf-service/src/work_journal.rs` and + `docs/spec/system_work_journal_v1.md`. +- Consumers: Work Journal create/readback, Memory Authority promotion checks, and + Dreaming Review integrations. +- Bump rule: Introduce a new identifier only when authority-allowance or accepted + promotion-reference semantics become incompatible. + ### Recall debug compact replay schema - Identifier: `elf.recall_debug.compact_replay/v1`. diff --git a/docs/spec/system_work_journal_v1.md b/docs/spec/system_work_journal_v1.md new file mode 100644 index 00000000..12b74885 --- /dev/null +++ b/docs/spec/system_work_journal_v1.md @@ -0,0 +1,173 @@ +--- +type: Spec +title: "Work Journal v1 Specification" +description: "Define source-adjacent Work Journal capture and readback without authoritative memory promotion." +resource: docs/spec/system_work_journal_v1.md +status: active +authority: normative +owner: spec +last_verified: 2026-06-27 +tags: + - docs + - spec + - work-journal +source_refs: + - https://linear.app/hackink/issue/XY-1117 +code_refs: + - packages/elf-service/src/work_journal.rs + - packages/elf-storage/src/work_journal.rs + - sql/tables/042_work_journal_entries.sql + - apps/elf-api/src/routes.rs + - apps/elf-mcp/src/server.rs +related: + - docs/spec/system_elf_memory_service_v2.md + - docs/spec/system_consolidation_proposals_v1.md + - docs/spec/system_recall_debug_panel_v1.md +drift_watch: + - packages/elf-service/src/work_journal.rs + - sql/tables/042_work_journal_entries.sql + - apps/elf-api/src/routes.rs + - apps/elf-mcp/src/server.rs +--- +# Work Journal v1 Specification + +Purpose: Define source-adjacent Work Journal capture and readback without authoritative memory promotion. +Status: normative +Read this when: You are implementing, validating, or using Work Journal session logs, handoff briefs, janitor reports, next-step captures, rejected-option captures, or readback. +Not this document: Memory Note authority, Source Library document indexing, Dreaming proposal review, or benchmark scoreboard contracts. +Defines: Work Journal v1 storage, entry families, redaction, source refs, promotion boundary, and readback behavior. + +## Boundary + +Work Journal is source-adjacent session evidence. It is not Memory Authority, not +Knowledge Workspace output, and not an archival search source. + +Work Journal rows must not: + +- create or update `memory_notes`; +- write `indexing_outbox`; +- create Qdrant points; +- answer current-fact questions as authoritative memory unless the response also + carries an accepted Memory Authority or Dreaming Review promotion reference. + +## Storage + +The source-of-truth table is `work_journal_entries`. + +Required row fields: + +- `entry_id`: stable UUID for the journal entry. +- `tenant_id`, `project_id`, `agent_id`, `scope`: normal ELF ownership and visibility context. +- `session_id`: stable session grouping key. +- `family`: one of the canonical entry families below. +- `status`: lifecycle state; v1 current readback uses `active`. +- `title`: optional operator-facing label. +- `body`: durable redacted journal text. +- `source_refs`: non-empty JSON array of non-empty JSON object source references supporting the entry. +- `explicit_next_steps`: JSON array of source-stated next steps. +- `inferred_next_steps`: JSON array of non-authoritative inferred next-step hints. +- `rejected_options`: JSON array of rejected options. +- `promotion_boundary`: normalized promotion metadata. +- `redaction_audit`: write-policy audit for the stored body. +- `created_at`, `updated_at`: timestamps. + +The table enforces storage-level checks for canonical `scope`, `family`, and `status` +values plus JSON shape checks for `source_refs`, side lists, `promotion_boundary`, +and `redaction_audit`. + +## Entry Families + +Canonical `family` values: + +- `session_log` +- `handoff_brief` +- `janitor_report` +- `explicit_next_step` +- `inferred_next_step` +- `rejected_option` + +Equivalent UI labels may be displayed, but API, MCP, storage, tests, and benchmark +fixtures must use the canonical values above. + +## Write Rules + +`POST /v2/work-journal/entries` captures one entry. + +Request-controlled fields: + +- `entry_id` optional UUID. When omitted, the service creates one. +- `scope`, `session_id`, `family`, `title`, `body`, `source_refs`. +- `write_policy` with the same exclusion/redaction shape used by note and document writes. +- `explicit_next_steps`, `inferred_next_steps`, `rejected_options`. +- `promotion_boundary`. + +Rules: + +- `source_refs` must be a non-empty JSON array of non-empty JSON objects. +- The English gate applies to body, title, list text, and identifier-like source-ref strings. +- `write_policy` is applied before persistence. +- If durable `body` or list text still contains secret markers after write-policy application, + the request is rejected. +- Shared-scope rows use normal ELF shared-grant behavior for readback. + +## Promotion Boundary + +The normalized `promotion_boundary` object uses schema +`elf.work_journal.promotion_boundary/v1`. + +Required normalized fields: + +- `schema = "elf.work_journal.promotion_boundary/v1"` +- `journal_entry_authority = "source_adjacent_only"` +- `authoritative_memory_allowed` +- `promotion_required_for_current_facts` +- `accepted_memory_authority_ref` +- `accepted_dreaming_review_ref` +- `requested_authoritative_memory_allowed` + +`authoritative_memory_allowed` may be true only when either +`accepted_memory_authority_ref` or `accepted_dreaming_review_ref` is present and +matches a supported accepted-reference shape. +If a caller requests authority without accepted promotion evidence, readback must preserve +that request in `requested_authoritative_memory_allowed` while keeping +`authoritative_memory_allowed = false`. + +In v1, supported accepted-reference shapes are intentionally narrow: + +- `accepted_memory_authority_ref`: JSON object with + `schema = "elf.memory_record_ref/v1"`, `kind = "note"`, UUID `id`, and + `status = "active"`. +- `accepted_dreaming_review_ref`: JSON object with + `schema = "elf.dreaming_review_queue/v1"`, UUID `proposal_id`, and + `review_state` of `approved` or `applied`. + +Primitive values, empty objects, wrong schemas, and incomplete objects are invalid +accepted references and must not make Work Journal authoritative. +Syntactically valid accepted references are not sufficient by themselves. The service +must resolve `accepted_memory_authority_ref` to an active, readable Memory Authority +note and `accepted_dreaming_review_ref` to an existing same-tenant/project Dreaming +Review or consolidation proposal in `approved` or `applied` state before setting +`authoritative_memory_allowed = true`. + +## Readback + +HTTP routes: + +- `GET /v2/work-journal/entries/{entry_id}` +- `POST /v2/work-journal/readback` + +MCP tools: + +- `elf_work_journal_entry_create` +- `elf_work_journal_entry_get` +- `elf_work_journal_session_readback` + +Session readback returns `elf.work_journal/v1` with newest-first `items` and an optional +`where_stopped` projection. `where_stopped` includes the latest entry id and family, +latest source refs, most recent returned explicit next steps, most recent returned +inferred next steps, most recent returned rejected options, and the latest promotion +boundary. + +Current-fact answers must route through accepted memory, graph, knowledge, or reviewed +Dreaming authority. Work Journal readback may answer "where did we stop?" with journal +evidence, but journal-only content remains source-adjacent. diff --git a/packages/elf-service/src/lib.rs b/packages/elf-service/src/lib.rs index 056200cc..8097224d 100644 --- a/packages/elf-service/src/lib.rs +++ b/packages/elf-service/src/lib.rs @@ -27,6 +27,7 @@ pub mod sharing; pub mod structured_fields; pub mod time_serde; pub mod update; +pub mod work_journal; mod access; mod error; @@ -146,6 +147,12 @@ pub use self::{ }, structured_fields::StructuredFields, update::{UpdateRequest, UpdateResponse}, + work_journal::{ + ELF_WORK_JOURNAL_SCHEMA_V1, WorkJournalEntryCreateRequest, WorkJournalEntryCreateResponse, + WorkJournalEntryFamily, WorkJournalEntryGetRequest, WorkJournalEntryResponse, + WorkJournalSessionReadbackRequest, WorkJournalSessionReadbackResponse, + WorkJournalWhereStopped, + }, }; use std::{future::Future, pin::Pin, sync::Arc}; diff --git a/packages/elf-service/src/work_journal.rs b/packages/elf-service/src/work_journal.rs new file mode 100644 index 00000000..e6c0eec5 --- /dev/null +++ b/packages/elf-service/src/work_journal.rs @@ -0,0 +1,1153 @@ +//! Source-adjacent Work Journal capture and readback APIs. + +use std::collections::HashSet; + +use serde::{Deserialize, Serialize}; +use serde_json::{self, Map, Value}; +use sqlx::PgConnection; +use time::OffsetDateTime; +use uuid::Uuid; + +use crate::{ + ElfService, Error, Result, + access::{self, ORG_PROJECT_ID, SharedSpaceGrantKey}, + search, +}; +use elf_config::Config; +use elf_domain::{ + english_gate, + writegate::{self, WritePolicy, WritePolicyAudit}, +}; +use elf_storage::{ + consolidation, + models::{MemoryNote, WorkJournalEntry}, + work_journal, +}; + +/// Schema identifier for Work Journal readback. +pub const ELF_WORK_JOURNAL_SCHEMA_V1: &str = "elf.work_journal/v1"; + +const WORK_JOURNAL_PROMOTION_BOUNDARY_SCHEMA_V1: &str = "elf.work_journal.promotion_boundary/v1"; +const DEFAULT_SESSION_READBACK_LIMIT: u32 = 20; +const MAX_SESSION_READBACK_LIMIT: u32 = 100; +const MAX_STORAGE_SCAN_ROWS: i64 = 500; +const MAX_BODY_CHARS: usize = 16_384; +const MAX_SIDE_LIST_ITEMS: usize = 64; + +/// Work Journal entry family. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Deserialize, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum WorkJournalEntryFamily { + /// Session log captured alongside source work. + SessionLog, + /// Handoff brief for another agent or future session. + HandoffBrief, + /// Janitor or cleanup report. + JanitorReport, + /// Explicit next step stated in the source. + ExplicitNextStep, + /// Inferred next step retained as a non-authoritative hint. + InferredNextStep, + /// Option that was considered and rejected. + RejectedOption, +} +impl WorkJournalEntryFamily { + /// Returns the canonical API/storage string. + pub fn as_str(self) -> &'static str { + match self { + Self::SessionLog => "session_log", + Self::HandoffBrief => "handoff_brief", + Self::JanitorReport => "janitor_report", + Self::ExplicitNextStep => "explicit_next_step", + Self::InferredNextStep => "inferred_next_step", + Self::RejectedOption => "rejected_option", + } + } + + fn parse(raw: &str) -> Result { + match raw { + "session_log" => Ok(Self::SessionLog), + "handoff_brief" => Ok(Self::HandoffBrief), + "janitor_report" => Ok(Self::JanitorReport), + "explicit_next_step" => Ok(Self::ExplicitNextStep), + "inferred_next_step" => Ok(Self::InferredNextStep), + "rejected_option" => Ok(Self::RejectedOption), + _ => Err(Error::InvalidRequest { + message: "family must be one of: session_log, handoff_brief, janitor_report, explicit_next_step, inferred_next_step, rejected_option.".to_string(), + }), + } + } +} + +impl ElfService { + /// Captures one source-adjacent Work Journal entry. + pub async fn work_journal_entry_create( + &self, + req: WorkJournalEntryCreateRequest, + ) -> Result { + let mut validated = validate_work_journal_create(&self.cfg, &req)?; + let now = OffsetDateTime::now_utc(); + let effective_project_id = if validated.scope == "org_shared" { + ORG_PROJECT_ID.to_string() + } else { + req.project_id.trim().to_string() + }; + let mut tx = self.db.pool.begin().await?; + + validated.promotion_boundary = resolve_promotion_boundary_authority( + &mut tx, + &self.cfg, + validated.promotion_boundary, + req.tenant_id.trim(), + req.project_id.trim(), + req.agent_id.trim(), + now, + ) + .await?; + + let entry = WorkJournalEntry { + entry_id: validated.entry_id, + tenant_id: req.tenant_id.trim().to_string(), + project_id: effective_project_id.clone(), + agent_id: req.agent_id.trim().to_string(), + scope: validated.scope, + session_id: validated.session_id, + family: req.family.as_str().to_string(), + status: "active".to_string(), + title: validated.title, + body: validated.body, + source_refs: validated.source_refs, + explicit_next_steps: validated.explicit_next_steps, + inferred_next_steps: validated.inferred_next_steps, + rejected_options: validated.rejected_options, + promotion_boundary: validated.promotion_boundary, + redaction_audit: serde_json::to_value(validated.redaction_audit).map_err(|err| { + Error::InvalidRequest { message: format!("redaction audit is invalid: {err}") } + })?, + created_at: now, + updated_at: now, + }; + + work_journal::insert_work_journal_entry(&mut *tx, &entry).await?; + + if entry.scope != "agent_private" { + access::ensure_active_project_scope_grant( + &mut *tx, + entry.tenant_id.as_str(), + effective_project_id.as_str(), + entry.scope.as_str(), + entry.agent_id.as_str(), + ) + .await?; + } + + tx.commit().await?; + + Ok(WorkJournalEntryCreateResponse { entry: row_to_response(entry)? }) + } + + /// Reads one source-adjacent Work Journal entry. + pub async fn work_journal_entry_get( + &self, + req: WorkJournalEntryGetRequest, + ) -> Result { + validate_read_context( + req.tenant_id.as_str(), + req.project_id.as_str(), + req.agent_id.as_str(), + req.read_profile.as_str(), + )?; + + let allowed_scopes = + search::resolve_read_profile_scopes(&self.cfg, req.read_profile.trim())?; + let shared_grants = load_work_journal_shared_grants( + self, + req.tenant_id.trim(), + req.project_id.trim(), + req.agent_id.trim(), + &allowed_scopes, + ) + .await?; + let row = + work_journal::get_work_journal_entry(&self.db.pool, req.tenant_id.trim(), req.entry_id) + .await? + .ok_or_else(|| Error::NotFound { + message: "Work Journal entry not found.".to_string(), + })?; + + if row.project_id != req.project_id.trim() && row.project_id != ORG_PROJECT_ID { + return Err(Error::NotFound { message: "Work Journal entry not found.".to_string() }); + } + if !work_journal_read_allowed(&row, req.agent_id.trim(), &allowed_scopes, &shared_grants) { + return Err(Error::ScopeDenied { + message: "Work Journal entry is not readable by this agent.".to_string(), + }); + } + + row_to_response(row) + } + + /// Reads newest-first Work Journal entries for one session. + pub async fn work_journal_session_readback( + &self, + req: WorkJournalSessionReadbackRequest, + ) -> Result { + validate_read_context( + req.tenant_id.as_str(), + req.project_id.as_str(), + req.agent_id.as_str(), + req.read_profile.as_str(), + )?; + validate_identifier(req.session_id.as_str(), "$.session_id")?; + + let limit = req + .limit + .unwrap_or(DEFAULT_SESSION_READBACK_LIMIT) + .clamp(1, MAX_SESSION_READBACK_LIMIT); + let allowed_scopes = + search::resolve_read_profile_scopes(&self.cfg, req.read_profile.trim())?; + let shared_grants = load_work_journal_shared_grants( + self, + req.tenant_id.trim(), + req.project_id.trim(), + req.agent_id.trim(), + &allowed_scopes, + ) + .await?; + let family_filter = + req.families.iter().copied().collect::>(); + let rows = work_journal::list_work_journal_entries_for_session( + &self.db.pool, + req.tenant_id.trim(), + req.project_id.trim(), + ORG_PROJECT_ID, + req.session_id.trim(), + MAX_STORAGE_SCAN_ROWS, + ) + .await?; + let mut items = Vec::new(); + + for row in rows { + if !family_filter.is_empty() + && !family_filter.contains(&WorkJournalEntryFamily::parse(row.family.as_str())?) + { + continue; + } + if !work_journal_read_allowed( + &row, + req.agent_id.trim(), + &allowed_scopes, + &shared_grants, + ) { + continue; + } + + items.push(row_to_response(row)?); + + if items.len() >= limit as usize { + break; + } + } + + let where_stopped = build_where_stopped(&items); + + Ok(WorkJournalSessionReadbackResponse { + schema: ELF_WORK_JOURNAL_SCHEMA_V1.to_string(), + session_id: req.session_id.trim().to_string(), + items, + where_stopped, + }) + } +} + +/// Request payload for source-adjacent Work Journal capture. +#[derive(Clone, Debug, Deserialize)] +pub struct WorkJournalEntryCreateRequest { + /// Tenant that owns the entry. + pub tenant_id: String, + /// Project that owns the entry. + pub project_id: String, + /// Agent capturing the entry. + pub agent_id: String, + /// Optional caller-supplied stable identifier. + pub entry_id: Option, + /// Visibility scope for readback. + pub scope: String, + /// Stable session identifier for grouping entries. + pub session_id: String, + /// Entry family. + pub family: WorkJournalEntryFamily, + /// Optional display title. + pub title: Option, + /// Journal body. This is source-adjacent, not authoritative memory. + pub body: String, + /// Source refs that support the journal entry. + pub source_refs: Vec, + /// Redaction/exclusion policy applied before persistence. + pub write_policy: Option, + #[serde(default)] + /// Explicit next steps stated by the captured source. + pub explicit_next_steps: Vec, + #[serde(default)] + /// Inferred next steps retained as non-authoritative hints. + pub inferred_next_steps: Vec, + #[serde(default)] + /// Options considered and rejected during the captured work. + pub rejected_options: Vec, + #[serde(default = "empty_object")] + /// Promotion boundary metadata. + pub promotion_boundary: Value, +} + +/// Response payload after Work Journal capture. +#[derive(Clone, Debug, Serialize)] +pub struct WorkJournalEntryCreateResponse { + /// Stored Work Journal entry. + pub entry: WorkJournalEntryResponse, +} + +/// Request payload for one Work Journal entry lookup. +#[derive(Clone, Debug, Deserialize)] +pub struct WorkJournalEntryGetRequest { + /// Tenant that owns the entry. + pub tenant_id: String, + /// Project used for read-profile and shared-grant checks. + pub project_id: String, + /// Agent requesting the read. + pub agent_id: String, + /// Read profile that determines visible scopes. + pub read_profile: String, + /// Entry identifier. + pub entry_id: Uuid, +} + +/// Request payload for session-level Work Journal readback. +#[derive(Clone, Debug, Deserialize)] +pub struct WorkJournalSessionReadbackRequest { + /// Tenant that owns the session. + pub tenant_id: String, + /// Project used for read-profile and shared-grant checks. + pub project_id: String, + /// Agent requesting the read. + pub agent_id: String, + /// Read profile that determines visible scopes. + pub read_profile: String, + /// Stable session identifier to read. + pub session_id: String, + #[serde(default)] + /// Optional family filter. + pub families: Vec, + /// Maximum number of returned entries. + pub limit: Option, +} + +/// Session-level Work Journal readback. +#[derive(Clone, Debug, Serialize)] +pub struct WorkJournalSessionReadbackResponse { + /// Readback schema identifier. + pub schema: String, + /// Stable session identifier. + pub session_id: String, + /// Newest-first journal entries. + pub items: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + /// Compact "where did we stop" projection from the returned entries. + pub where_stopped: Option, +} + +/// One source-adjacent Work Journal entry returned by readback. +#[derive(Clone, Debug, Serialize)] +pub struct WorkJournalEntryResponse { + /// Readback schema identifier. + pub schema: String, + /// Journal entry identifier. + pub entry_id: Uuid, + /// Tenant that owns the entry. + pub tenant_id: String, + /// Project that owns the entry. + pub project_id: String, + /// Agent that captured the entry. + pub agent_id: String, + /// Visibility scope for readback. + pub scope: String, + /// Stable session identifier. + pub session_id: String, + /// Entry family. + pub family: WorkJournalEntryFamily, + /// Lifecycle status. + pub status: String, + #[serde(skip_serializing_if = "Option::is_none")] + /// Optional display title. + pub title: Option, + /// Redacted durable journal body. + pub body: String, + /// Source refs supporting the entry. + pub source_refs: Vec, + /// Explicit next steps stated by the captured source. + pub explicit_next_steps: Vec, + /// Inferred next steps retained as non-authoritative hints. + pub inferred_next_steps: Vec, + /// Rejected options captured by the journal. + pub rejected_options: Vec, + /// Promotion boundary metadata. + pub promotion_boundary: Value, + /// Redaction audit for the durable journal body. + pub redaction_audit: WritePolicyAudit, + /// Creation timestamp. + pub created_at: OffsetDateTime, + /// Last update timestamp. + pub updated_at: OffsetDateTime, +} + +/// Compact "where did we stop" projection for one journal session. +#[derive(Clone, Debug, Serialize)] +pub struct WorkJournalWhereStopped { + /// Latest returned entry identifier. + pub latest_entry_id: Uuid, + /// Latest returned entry family. + pub latest_family: WorkJournalEntryFamily, + /// Source refs associated with the latest returned entry. + pub source_refs: Vec, + /// Most recent explicit next steps in returned entries. + pub explicit_next_steps: Vec, + /// Most recent inferred next steps in returned entries. + pub inferred_next_steps: Vec, + /// Most recent rejected options in returned entries. + pub rejected_options: Vec, + /// Promotion boundary for the latest returned entry. + pub promotion_boundary: Value, +} + +struct ValidatedWorkJournalCreate { + entry_id: Uuid, + scope: String, + session_id: String, + title: Option, + body: String, + source_refs: Value, + explicit_next_steps: Value, + inferred_next_steps: Value, + rejected_options: Value, + promotion_boundary: Value, + redaction_audit: WritePolicyAudit, +} + +fn validate_work_journal_create( + cfg: &Config, + req: &WorkJournalEntryCreateRequest, +) -> Result { + validate_write_context( + cfg, + req.tenant_id.as_str(), + req.project_id.as_str(), + req.agent_id.as_str(), + req.scope.as_str(), + )?; + validate_identifier(req.session_id.as_str(), "$.session_id")?; + + if req.body.trim().is_empty() { + return Err(Error::InvalidRequest { message: "body must be non-empty.".to_string() }); + } + if req.body.chars().count() > MAX_BODY_CHARS { + return Err(Error::InvalidRequest { + message: "body exceeds max journal size.".to_string(), + }); + } + + let title = req.title.as_ref().map(|title| title.trim().to_string()).filter(|s| !s.is_empty()); + + if let Some(title) = title.as_ref() { + validate_natural_language(title.as_str(), "$.title")?; + + if writegate::contains_secrets(title.as_str()) { + return Err(Error::InvalidRequest { message: "title contains secrets.".to_string() }); + } + } + + let policy_result = writegate::apply_write_policy(req.body.as_str(), req.write_policy.as_ref()) + .map_err(|err| Error::InvalidRequest { + message: format!("write_policy is invalid: {err:?}"), + })?; + let body = policy_result.transformed; + + if body.trim().is_empty() { + return Err(Error::InvalidRequest { message: "body must be non-empty.".to_string() }); + } + + validate_natural_language(body.as_str(), "$.body")?; + + if writegate::contains_secrets(body.as_str()) { + return Err(Error::InvalidRequest { message: "body contains secrets.".to_string() }); + } + + validate_text_list(&req.explicit_next_steps, "$.explicit_next_steps")?; + validate_text_list(&req.inferred_next_steps, "$.inferred_next_steps")?; + validate_text_list(&req.rejected_options, "$.rejected_options")?; + + let source_refs = validate_source_refs(&req.source_refs)?; + let promotion_boundary = normalize_promotion_boundary(&req.promotion_boundary)?; + let explicit_next_steps = serde_json::to_value(&req.explicit_next_steps).map_err(|err| { + Error::InvalidRequest { message: format!("explicit_next_steps are invalid: {err}") } + })?; + let inferred_next_steps = serde_json::to_value(&req.inferred_next_steps).map_err(|err| { + Error::InvalidRequest { message: format!("inferred_next_steps are invalid: {err}") } + })?; + let rejected_options = serde_json::to_value(&req.rejected_options).map_err(|err| { + Error::InvalidRequest { message: format!("rejected_options are invalid: {err}") } + })?; + + Ok(ValidatedWorkJournalCreate { + entry_id: req.entry_id.unwrap_or_else(Uuid::new_v4), + scope: req.scope.trim().to_string(), + session_id: req.session_id.trim().to_string(), + title, + body, + source_refs, + explicit_next_steps, + inferred_next_steps, + rejected_options, + promotion_boundary, + redaction_audit: policy_result.audit, + }) +} + +fn validate_write_context( + cfg: &Config, + tenant_id: &str, + project_id: &str, + agent_id: &str, + scope: &str, +) -> Result<()> { + if tenant_id.trim().is_empty() + || project_id.trim().is_empty() + || agent_id.trim().is_empty() + || scope.trim().is_empty() + { + return Err(Error::InvalidRequest { + message: "tenant_id, project_id, agent_id, and scope are required.".to_string(), + }); + } + + validate_identifier(tenant_id, "$.tenant_id")?; + validate_identifier(project_id, "$.project_id")?; + validate_identifier(agent_id, "$.agent_id")?; + + if !cfg.scopes.allowed.iter().any(|allowed| allowed == scope.trim()) { + return Err(Error::ScopeDenied { message: "scope is not allowed.".to_string() }); + } + if !scope_write_allowed(cfg, scope.trim()) { + return Err(Error::ScopeDenied { message: "scope is not writable.".to_string() }); + } + + Ok(()) +} + +fn validate_read_context( + tenant_id: &str, + project_id: &str, + agent_id: &str, + read_profile: &str, +) -> Result<()> { + if tenant_id.trim().is_empty() + || project_id.trim().is_empty() + || agent_id.trim().is_empty() + || read_profile.trim().is_empty() + { + return Err(Error::InvalidRequest { + message: "tenant_id, project_id, agent_id, and read_profile are required.".to_string(), + }); + } + + validate_identifier(tenant_id, "$.tenant_id")?; + validate_identifier(project_id, "$.project_id")?; + validate_identifier(agent_id, "$.agent_id")?; + validate_identifier(read_profile, "$.read_profile")?; + + Ok(()) +} + +fn validate_text_list(values: &[String], path: &str) -> Result<()> { + if values.len() > MAX_SIDE_LIST_ITEMS { + return Err(Error::InvalidRequest { message: format!("{path} has too many items.") }); + } + + for (index, value) in values.iter().enumerate() { + if value.trim().is_empty() { + return Err(Error::InvalidRequest { + message: format!("{path}[{index}] must be non-empty."), + }); + } + + validate_natural_language(value.as_str(), format!("{path}[{index}]").as_str())?; + + if writegate::contains_secrets(value.as_str()) { + return Err(Error::InvalidRequest { + message: format!("{path}[{index}] contains secrets."), + }); + } + } + + Ok(()) +} + +fn validate_source_refs(source_refs: &[Value]) -> Result { + if source_refs.is_empty() { + return Err(Error::InvalidRequest { + message: "source_refs must be non-empty.".to_string(), + }); + } + if source_refs.len() > MAX_SIDE_LIST_ITEMS { + return Err(Error::InvalidRequest { + message: "source_refs has too many items.".to_string(), + }); + } + + for (index, source_ref) in source_refs.iter().enumerate() { + match source_ref { + Value::Object(map) if !map.is_empty() => {}, + _ => { + return Err(Error::InvalidRequest { + message: format!("source_refs[{index}] must be a non-empty object."), + }); + }, + } + } + + let value = Value::Array(source_refs.to_vec()); + + validate_json_strings(&value, "$.source_refs")?; + + Ok(value) +} + +fn validate_json_strings(value: &Value, path: &str) -> Result<()> { + match value { + Value::String(text) => { + validate_identifier(text.as_str(), path)?; + + if writegate::contains_secrets(text.as_str()) { + return Err(Error::InvalidRequest { message: format!("{path} contains secrets.") }); + } + }, + Value::Array(items) => + for (index, item) in items.iter().enumerate() { + validate_json_strings(item, format!("{path}[{index}]").as_str())?; + }, + Value::Object(map) => + for (key, item) in map { + validate_identifier(key.as_str(), format!("{path}.{key}").as_str())?; + validate_json_strings(item, format!("{path}.{key}").as_str())?; + }, + Value::Null | Value::Bool(_) | Value::Number(_) => {}, + } + + Ok(()) +} + +fn normalize_promotion_boundary(input: &Value) -> Result { + let map = match input { + Value::Null => Map::new(), + Value::Object(map) => map.clone(), + _ => { + return Err(Error::InvalidRequest { + message: "promotion_boundary must be a JSON object.".to_string(), + }); + }, + }; + + validate_json_strings(&Value::Object(map.clone()), "$.promotion_boundary")?; + + let accepted_memory_authority_ref = map.get("accepted_memory_authority_ref").cloned(); + let accepted_dreaming_review_ref = map.get("accepted_dreaming_review_ref").cloned(); + + if accepted_memory_authority_ref + .as_ref() + .is_some_and(|value| !value.is_null() && !is_valid_memory_authority_ref(value)) + { + return Err(Error::InvalidRequest { + message: + "accepted_memory_authority_ref must be an active elf.memory_record_ref/v1 note ref." + .to_string(), + }); + } + if accepted_dreaming_review_ref + .as_ref() + .is_some_and(|value| !value.is_null() && !is_valid_dreaming_review_ref(value)) + { + return Err(Error::InvalidRequest { + message: + "accepted_dreaming_review_ref must be an accepted elf.dreaming_review_queue/v1 ref." + .to_string(), + }); + } + + Ok(serde_json::json!({ + "schema": WORK_JOURNAL_PROMOTION_BOUNDARY_SCHEMA_V1, + "journal_entry_authority": "source_adjacent_only", + "authoritative_memory_allowed": false, + "promotion_required_for_current_facts": true, + "accepted_memory_authority_ref": accepted_memory_authority_ref.unwrap_or(Value::Null), + "accepted_dreaming_review_ref": accepted_dreaming_review_ref.unwrap_or(Value::Null), + "requested_authoritative_memory_allowed": map + .get("authoritative_memory_allowed") + .and_then(Value::as_bool) + .unwrap_or(false), + })) +} + +fn is_valid_memory_authority_ref(value: &Value) -> bool { + let Some(map) = value.as_object() else { + return false; + }; + let Some(id) = object_string(map, "id") else { + return false; + }; + + object_string(map, "schema") == Some("elf.memory_record_ref/v1") + && object_string(map, "kind") == Some("note") + && object_string(map, "status") == Some("active") + && Uuid::parse_str(id).is_ok() +} + +fn memory_ref_id(value: &Value) -> Option { + Uuid::parse_str(object_string(value.as_object()?, "id")?).ok() +} + +fn is_valid_dreaming_review_ref(value: &Value) -> bool { + let Some(map) = value.as_object() else { + return false; + }; + let Some(proposal_id) = object_string(map, "proposal_id") else { + return false; + }; + let review_state = object_string(map, "review_state"); + + object_string(map, "schema") == Some("elf.dreaming_review_queue/v1") + && Uuid::parse_str(proposal_id).is_ok() + && matches!(review_state, Some("approved" | "applied")) +} + +fn dreaming_ref_proposal_id(value: &Value) -> Option { + Uuid::parse_str(object_string(value.as_object()?, "proposal_id")?).ok() +} + +fn object_string<'a>(map: &'a Map, key: &str) -> Option<&'a str> { + map.get(key).and_then(Value::as_str).map(str::trim).filter(|value| !value.is_empty()) +} + +fn validate_identifier(text: &str, field: &str) -> Result<()> { + if text.trim().is_empty() || !english_gate::is_english_identifier(text.trim()) { + return Err(Error::NonEnglishInput { field: field.to_string() }); + } + + Ok(()) +} + +fn validate_natural_language(text: &str, field: &str) -> Result<()> { + if !english_gate::is_english_natural_language(text) { + return Err(Error::NonEnglishInput { field: field.to_string() }); + } + + Ok(()) +} + +fn scope_write_allowed(cfg: &Config, scope: &str) -> bool { + match scope { + "agent_private" => cfg.scopes.write_allowed.agent_private, + "project_shared" => cfg.scopes.write_allowed.project_shared, + "org_shared" => cfg.scopes.write_allowed.org_shared, + _ => false, + } +} + +fn work_journal_read_allowed( + entry: &WorkJournalEntry, + requester_agent_id: &str, + allowed_scopes: &[String], + shared_grants: &HashSet, +) -> bool { + if entry.status != "active" { + return false; + } + if !allowed_scopes.iter().any(|scope| scope == &entry.scope) { + return false; + } + if entry.scope == "agent_private" { + return entry.agent_id == requester_agent_id; + } + if entry.agent_id == requester_agent_id { + return true; + } + + shared_grants.contains(&SharedSpaceGrantKey { + scope: entry.scope.clone(), + space_owner_agent_id: entry.agent_id.clone(), + }) +} + +fn row_to_response(row: WorkJournalEntry) -> Result { + let family = WorkJournalEntryFamily::parse(row.family.as_str())?; + let redaction_audit = serde_json::from_value::(row.redaction_audit.clone()) + .map_err(|err| Error::InvalidRequest { + message: format!("stored redaction audit is invalid: {err}"), + })?; + + Ok(WorkJournalEntryResponse { + schema: ELF_WORK_JOURNAL_SCHEMA_V1.to_string(), + entry_id: row.entry_id, + tenant_id: row.tenant_id, + project_id: row.project_id, + agent_id: row.agent_id, + scope: row.scope, + session_id: row.session_id, + family, + status: row.status, + title: row.title, + body: row.body, + source_refs: value_array(row.source_refs), + explicit_next_steps: string_array(row.explicit_next_steps), + inferred_next_steps: string_array(row.inferred_next_steps), + rejected_options: string_array(row.rejected_options), + promotion_boundary: row.promotion_boundary, + redaction_audit, + created_at: row.created_at, + updated_at: row.updated_at, + }) +} + +fn value_array(value: Value) -> Vec { + match value { + Value::Array(items) => items, + _ => Vec::new(), + } +} + +fn string_array(value: Value) -> Vec { + match value { + Value::Array(items) => + items.into_iter().filter_map(|item| item.as_str().map(str::to_string)).collect(), + _ => Vec::new(), + } +} + +fn build_where_stopped(items: &[WorkJournalEntryResponse]) -> Option { + let latest = items.first()?; + let explicit_next_steps = first_non_empty(items.iter().map(|item| &item.explicit_next_steps)); + let inferred_next_steps = first_non_empty(items.iter().map(|item| &item.inferred_next_steps)); + let rejected_options = first_non_empty(items.iter().map(|item| &item.rejected_options)); + + Some(WorkJournalWhereStopped { + latest_entry_id: latest.entry_id, + latest_family: latest.family, + source_refs: latest.source_refs.clone(), + explicit_next_steps, + inferred_next_steps, + rejected_options, + promotion_boundary: latest.promotion_boundary.clone(), + }) +} + +fn first_non_empty<'a>(mut lists: impl Iterator>) -> Vec { + lists.find(|items| !items.is_empty()).cloned().unwrap_or_default() +} + +fn empty_object() -> Value { + Value::Object(Map::new()) +} + +async fn resolve_promotion_boundary_authority( + executor: &mut PgConnection, + cfg: &Config, + mut boundary: Value, + tenant_id: &str, + project_id: &str, + agent_id: &str, + now: OffsetDateTime, +) -> Result { + let memory_ref = boundary.get("accepted_memory_authority_ref").cloned(); + let dreaming_ref = boundary.get("accepted_dreaming_review_ref").cloned(); + let mut has_accepted_ref = false; + + if let Some(memory_ref) = + memory_ref.as_ref().filter(|value| is_valid_memory_authority_ref(value)) + { + if !accepted_memory_authority_ref_is_readable( + &mut *executor, + cfg, + memory_ref, + tenant_id, + project_id, + agent_id, + now, + ) + .await? + { + return Err(Error::InvalidRequest { + message: "accepted_memory_authority_ref was not found or is not readable." + .to_string(), + }); + } + + has_accepted_ref = true; + } + if let Some(dreaming_ref) = + dreaming_ref.as_ref().filter(|value| is_valid_dreaming_review_ref(value)) + { + if !accepted_dreaming_review_ref_exists(&mut *executor, dreaming_ref, tenant_id, project_id) + .await? + { + return Err(Error::InvalidRequest { + message: "accepted_dreaming_review_ref was not found or is not accepted." + .to_string(), + }); + } + + has_accepted_ref = true; + } + + boundary["authoritative_memory_allowed"] = Value::Bool(has_accepted_ref); + boundary["promotion_required_for_current_facts"] = Value::Bool(!has_accepted_ref); + + Ok(boundary) +} + +async fn accepted_memory_authority_ref_is_readable( + executor: &mut PgConnection, + cfg: &Config, + memory_ref: &Value, + tenant_id: &str, + project_id: &str, + agent_id: &str, + now: OffsetDateTime, +) -> Result { + let Some(note_id) = memory_ref_id(memory_ref) else { + return Ok(false); + }; + let note = sqlx::query_as::<_, MemoryNote>( + "\ +SELECT * +FROM memory_notes +WHERE note_id = $1 + AND tenant_id = $2 + AND project_id IN ($3, $4) +LIMIT 1", + ) + .bind(note_id) + .bind(tenant_id) + .bind(project_id) + .bind(ORG_PROJECT_ID) + .fetch_optional(&mut *executor) + .await?; + let Some(note) = note else { + return Ok(false); + }; + let org_shared_allowed = cfg.scopes.allowed.iter().any(|scope| scope == "org_shared"); + let shared_grants = access::load_shared_read_grants_with_org_shared( + &mut *executor, + tenant_id, + project_id, + agent_id, + org_shared_allowed, + ) + .await?; + + Ok(access::note_read_allowed(¬e, agent_id, &cfg.scopes.allowed, &shared_grants, now)) +} + +async fn accepted_dreaming_review_ref_exists( + executor: &mut PgConnection, + dreaming_ref: &Value, + tenant_id: &str, + project_id: &str, +) -> Result { + let Some(proposal_id) = dreaming_ref_proposal_id(dreaming_ref) else { + return Ok(false); + }; + let Some(proposal) = consolidation::get_consolidation_proposal( + &mut *executor, + tenant_id, + project_id, + proposal_id, + ) + .await? + else { + return Ok(false); + }; + let Some(map) = dreaming_ref.as_object() else { + return Ok(false); + }; + + Ok(matches!(proposal.review_state.as_str(), "approved" | "applied") + && object_string(map, "review_state") == Some(proposal.review_state.as_str())) +} + +async fn load_work_journal_shared_grants( + service: &ElfService, + tenant_id: &str, + project_id: &str, + agent_id: &str, + allowed_scopes: &[String], +) -> Result> { + let org_shared_allowed = allowed_scopes.iter().any(|scope| scope == "org_shared"); + + access::load_shared_read_grants_with_org_shared( + &service.db.pool, + tenant_id, + project_id, + agent_id, + org_shared_allowed, + ) + .await +} + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + + use serde_json; + use time::OffsetDateTime; + use uuid::Uuid; + + use crate::{ + access::SharedSpaceGrantKey, + work_journal::{self, WORK_JOURNAL_PROMOTION_BOUNDARY_SCHEMA_V1}, + }; + use elf_storage::models::WorkJournalEntry; + + #[test] + fn promotion_boundary_flags_journal_only_without_accepted_ref() { + let boundary = work_journal::normalize_promotion_boundary(&serde_json::json!({ + "authoritative_memory_allowed": true + })) + .expect("boundary should normalize"); + + assert_eq!(boundary["schema"], WORK_JOURNAL_PROMOTION_BOUNDARY_SCHEMA_V1); + assert_eq!(boundary["authoritative_memory_allowed"], false); + assert_eq!(boundary["promotion_required_for_current_facts"], true); + assert_eq!(boundary["requested_authoritative_memory_allowed"], true); + } + + #[test] + fn promotion_boundary_preserves_memory_ref_without_granting_shape_only_authority() { + let boundary = work_journal::normalize_promotion_boundary(&serde_json::json!({ + "accepted_memory_authority_ref": { + "schema": "elf.memory_record_ref/v1", + "kind": "note", + "id": "11111111-1111-1111-1111-111111111111", + "status": "active" + } + })) + .expect("boundary should normalize"); + + assert_eq!(boundary["authoritative_memory_allowed"], false); + assert_eq!(boundary["promotion_required_for_current_facts"], true); + assert_eq!( + boundary["accepted_memory_authority_ref"]["id"], + serde_json::json!("11111111-1111-1111-1111-111111111111") + ); + } + + #[test] + fn promotion_boundary_preserves_dreaming_ref_without_granting_shape_only_authority() { + let boundary = work_journal::normalize_promotion_boundary(&serde_json::json!({ + "accepted_dreaming_review_ref": { + "schema": "elf.dreaming_review_queue/v1", + "proposal_id": "22222222-2222-4222-8222-222222222222", + "review_state": "applied" + } + })) + .expect("boundary should normalize"); + + assert_eq!(boundary["authoritative_memory_allowed"], false); + assert_eq!(boundary["promotion_required_for_current_facts"], true); + assert_eq!( + boundary["accepted_dreaming_review_ref"]["proposal_id"], + serde_json::json!("22222222-2222-4222-8222-222222222222") + ); + } + + #[test] + fn promotion_boundary_rejects_forged_accepted_refs() { + let primitive_result = work_journal::normalize_promotion_boundary(&serde_json::json!({ + "accepted_memory_authority_ref": true + })); + let object_result = work_journal::normalize_promotion_boundary(&serde_json::json!({ + "accepted_memory_authority_ref": { + "schema": "elf.memory_record_ref/v1", + "id": "11111111-1111-1111-1111-111111111111" + } + })); + + assert!(primitive_result.is_err()); + assert!(object_result.is_err()); + } + + #[test] + fn source_refs_reject_non_object_items() { + let result = work_journal::validate_source_refs(&[serde_json::json!("XY-1117")]); + + assert!(result.is_err()); + } + + #[test] + fn read_allowed_enforces_private_and_shared_grants() { + let allowed = vec!["agent_private".to_string(), "project_shared".to_string()]; + let no_grants = HashSet::new(); + let private = journal_row("agent_private", "agent-a", "active"); + let shared = journal_row("project_shared", "agent-a", "active"); + let inactive = journal_row("agent_private", "agent-a", "deleted"); + + assert!(work_journal::work_journal_read_allowed(&private, "agent-a", &allowed, &no_grants)); + assert!(!work_journal::work_journal_read_allowed( + &private, "agent-b", &allowed, &no_grants + )); + assert!(!work_journal::work_journal_read_allowed( + &inactive, "agent-a", &allowed, &no_grants + )); + assert!(work_journal::work_journal_read_allowed(&shared, "agent-a", &allowed, &no_grants)); + assert!(!work_journal::work_journal_read_allowed(&shared, "agent-b", &allowed, &no_grants)); + + let mut grants = HashSet::new(); + + grants.insert(SharedSpaceGrantKey { + scope: "project_shared".to_string(), + space_owner_agent_id: "agent-a".to_string(), + }); + + assert!(work_journal::work_journal_read_allowed(&shared, "agent-b", &allowed, &grants)); + + let private_only = vec!["agent_private".to_string()]; + + assert!(!work_journal::work_journal_read_allowed( + &shared, + "agent-b", + &private_only, + &grants + )); + } + + fn journal_row(scope: &str, agent_id: &str, status: &str) -> WorkJournalEntry { + let now = OffsetDateTime::now_utc(); + + WorkJournalEntry { + entry_id: Uuid::nil(), + tenant_id: "tenant".to_string(), + project_id: "project".to_string(), + agent_id: agent_id.to_string(), + scope: scope.to_string(), + session_id: "session".to_string(), + family: "session_log".to_string(), + status: status.to_string(), + title: None, + body: "body".to_string(), + source_refs: serde_json::json!([{ "schema": "source_ref/v1" }]), + explicit_next_steps: serde_json::json!([]), + inferred_next_steps: serde_json::json!([]), + rejected_options: serde_json::json!([]), + promotion_boundary: serde_json::json!({}), + redaction_audit: serde_json::json!({}), + created_at: now, + updated_at: now, + } + } +} diff --git a/packages/elf-service/tests/acceptance/suite.rs b/packages/elf-service/tests/acceptance/suite.rs index 7db8daac..e2b615ec 100644 --- a/packages/elf-service/tests/acceptance/suite.rs +++ b/packages/elf-service/tests/acceptance/suite.rs @@ -14,6 +14,7 @@ mod rebuild_qdrant; mod sot_vectors; mod structured_field_retrieval; mod trace_admin_observability; +mod work_journal; use std::{ env, fs, diff --git a/packages/elf-service/tests/acceptance/work_journal.rs b/packages/elf-service/tests/acceptance/work_journal.rs new file mode 100644 index 00000000..16d39c61 --- /dev/null +++ b/packages/elf-service/tests/acceptance/work_journal.rs @@ -0,0 +1,379 @@ +use std::sync::{Arc, atomic::AtomicUsize}; + +use serde_json::Value; +use time::OffsetDateTime; +use uuid::Uuid; + +use crate::acceptance::{self, SpyExtractor, StubEmbedding, StubRerank}; +use elf_domain::writegate::{WritePolicy, WriteRedaction, WriteSpan}; +use elf_service::{ + ElfService, Error, Providers, WorkJournalEntryCreateRequest, WorkJournalEntryFamily, + WorkJournalEntryGetRequest, WorkJournalSessionReadbackRequest, +}; +use elf_storage::{db::Db, qdrant::QdrantStore}; +use elf_testkit::TestDatabase; + +fn work_journal_entry_request(entry_id: Uuid) -> WorkJournalEntryCreateRequest { + WorkJournalEntryCreateRequest { + tenant_id: "tenant".to_string(), + project_id: "project".to_string(), + agent_id: "agent-a".to_string(), + entry_id: Some(entry_id), + scope: "agent_private".to_string(), + session_id: "xy-1117-session".to_string(), + family: WorkJournalEntryFamily::SessionLog, + title: Some("XY-1117 session log".to_string()), + body: "Work stopped after the dry run failed with api_key=abcdef123.".to_string(), + source_refs: vec![serde_json::json!({ + "schema": "source_ref/v1", + "resolver": "work_journal_test/v1", + "ref": { + "issue": "XY-1117", + "session_id": "xy-1117-session" + } + })], + write_policy: Some(WritePolicy { + exclusions: vec![], + redactions: vec![WriteRedaction::Replace { + span: WriteSpan { start: 43, end: 60 }, + replacement: "[redacted credential]".to_string(), + }], + }), + explicit_next_steps: vec!["Run the Work Journal validation tests.".to_string()], + inferred_next_steps: vec![ + "Keep journal evidence separate from current memory answers.".to_string(), + ], + rejected_options: vec![ + "Do not store this session log as an authoritative memory note.".to_string(), + ], + promotion_boundary: serde_json::json!({ "authoritative_memory_allowed": true }), + } +} + +fn request_with_promotion_boundary( + entry_id: Uuid, + promotion_boundary: Value, +) -> WorkJournalEntryCreateRequest { + let mut request = work_journal_entry_request(entry_id); + + request.body = "Work stopped after accepted promotion evidence was reviewed.".to_string(); + request.write_policy = None; + request.promotion_boundary = promotion_boundary; + + request +} + +fn memory_record_ref(note_id: Uuid) -> Value { + serde_json::json!({ + "schema": "elf.memory_record_ref/v1", + "kind": "note", + "id": note_id, + "status": "active" + }) +} + +fn dreaming_review_ref(proposal_id: Uuid, review_state: &str) -> Value { + serde_json::json!({ + "schema": "elf.dreaming_review_queue/v1", + "proposal_id": proposal_id, + "review_state": review_state + }) +} + +async fn work_journal_service() -> Option<(ElfService, TestDatabase)> { + let Some(dsn) = elf_testkit::env_dsn() else { + eprintln!("Skipping work_journal acceptance; set ELF_PG_DSN to run this test."); + + return None; + }; + let test_db = TestDatabase::new(&dsn).await.expect("Failed to create test database."); + let cfg = acceptance::test_config( + test_db.dsn().to_string(), + "http://127.0.0.1:1".to_string(), + 4_096, + test_db.collection_name("elf_acceptance_notes"), + test_db.collection_name("elf_acceptance_docs"), + ); + let db = Db::connect(&cfg.storage.postgres).await.expect("Failed to connect test DB."); + + db.ensure_schema(cfg.storage.qdrant.vector_dim).await.expect("Failed to ensure schema"); + + let qdrant = QdrantStore::new(&cfg.storage.qdrant).expect("Failed to build qdrant store"); + let providers = Providers::new( + Arc::new(StubEmbedding { vector_dim: 4_096 }), + Arc::new(StubRerank), + Arc::new(SpyExtractor { + calls: Arc::new(AtomicUsize::new(0)), + payload: serde_json::json!({}), + }), + ); + let service = ElfService::with_providers(cfg, db, qdrant, providers); + + Some((service, test_db)) +} + +#[tokio::test] +async fn work_journal_persists_redacted_source_adjacent_session_readback() { + let Some((service, test_db)) = work_journal_service().await else { + return; + }; + let entry_id = Uuid::parse_str("aaaaaaaa-1111-4111-8111-aaaaaaaa1111").expect("uuid"); + let created = service + .work_journal_entry_create(work_journal_entry_request(entry_id)) + .await + .expect("journal entry should persist"); + + assert_eq!(created.entry.entry_id, entry_id); + assert!(created.entry.body.contains("[redacted credential]")); + assert!(!created.entry.body.contains("abcdef123")); + assert_eq!( + created.entry.promotion_boundary["authoritative_memory_allowed"], + serde_json::json!(false) + ); + + let fetched = service + .work_journal_entry_get(WorkJournalEntryGetRequest { + tenant_id: "tenant".to_string(), + project_id: "project".to_string(), + agent_id: "agent-a".to_string(), + read_profile: "private_only".to_string(), + entry_id, + }) + .await + .expect("journal entry should be readable"); + + assert_eq!(fetched.entry_id, entry_id); + assert_eq!(fetched.source_refs.len(), 1); + + let readback = service + .work_journal_session_readback(WorkJournalSessionReadbackRequest { + tenant_id: "tenant".to_string(), + project_id: "project".to_string(), + agent_id: "agent-a".to_string(), + read_profile: "private_only".to_string(), + session_id: "xy-1117-session".to_string(), + families: vec![], + limit: Some(10), + }) + .await + .expect("session readback should load journal evidence"); + + assert_eq!(readback.items.len(), 1); + + let where_stopped = readback.where_stopped.expect("where_stopped should be present"); + + assert_eq!(where_stopped.latest_entry_id, entry_id); + assert_eq!( + where_stopped.explicit_next_steps, + vec!["Run the Work Journal validation tests.".to_string()] + ); + assert_eq!( + where_stopped.promotion_boundary["authoritative_memory_allowed"], + serde_json::json!(false) + ); + + let memory_count: i64 = sqlx::query_scalar("SELECT count(*) FROM memory_notes") + .fetch_one(&service.db.pool) + .await + .expect("memory_notes count should query"); + let outbox_count: i64 = sqlx::query_scalar("SELECT count(*) FROM indexing_outbox") + .fetch_one(&service.db.pool) + .await + .expect("indexing_outbox count should query"); + + assert_eq!(memory_count, 0, "Work Journal must not create authoritative memory notes"); + assert_eq!(outbox_count, 0, "Work Journal must not enqueue memory indexing"); + + test_db.cleanup().await.expect("Failed to cleanup test database."); +} + +#[tokio::test] +async fn work_journal_promotion_boundary_requires_existing_accepted_refs() { + let Some((service, test_db)) = work_journal_service().await else { + return; + }; + let forged_note_id = Uuid::parse_str("bbbbbbbb-1111-4111-8111-bbbbbbbb1111").expect("uuid"); + let forged_note_request = request_with_promotion_boundary( + Uuid::parse_str("bbbbbbbb-2222-4222-8222-bbbbbbbb2222").expect("uuid"), + serde_json::json!({ + "accepted_memory_authority_ref": memory_record_ref(forged_note_id), + }), + ); + let forged_note_error = service + .work_journal_entry_create(forged_note_request) + .await + .expect_err("syntactically valid but nonexistent memory authority ref should be rejected"); + + assert!(matches!( + forged_note_error, + Error::InvalidRequest { message } if message.contains("accepted_memory_authority_ref") + )); + + let accepted_note_id = Uuid::parse_str("cccccccc-1111-4111-8111-cccccccc1111").expect("uuid"); + + insert_active_memory_note(&service, accepted_note_id).await; + + let accepted_note_request = request_with_promotion_boundary( + Uuid::parse_str("cccccccc-2222-4222-8222-cccccccc2222").expect("uuid"), + serde_json::json!({ + "accepted_memory_authority_ref": memory_record_ref(accepted_note_id), + }), + ); + let accepted_note = service + .work_journal_entry_create(accepted_note_request) + .await + .expect("existing active memory authority ref should be accepted"); + + assert_eq!( + accepted_note.entry.promotion_boundary["authoritative_memory_allowed"], + serde_json::json!(true) + ); + + let forged_proposal_id = Uuid::parse_str("dddddddd-1111-4111-8111-dddddddd1111").expect("uuid"); + let forged_proposal_request = request_with_promotion_boundary( + Uuid::parse_str("dddddddd-2222-4222-8222-dddddddd2222").expect("uuid"), + serde_json::json!({ + "accepted_dreaming_review_ref": dreaming_review_ref(forged_proposal_id, "applied"), + }), + ); + let forged_proposal_error = service + .work_journal_entry_create(forged_proposal_request) + .await + .expect_err("syntactically valid but nonexistent dreaming review ref should be rejected"); + + assert!(matches!( + forged_proposal_error, + Error::InvalidRequest { message } if message.contains("accepted_dreaming_review_ref") + )); + + let accepted_proposal_id = + Uuid::parse_str("eeeeeeee-1111-4111-8111-eeeeeeee1111").expect("uuid"); + + insert_applied_dreaming_proposal(&service, accepted_proposal_id).await; + + let accepted_proposal_request = request_with_promotion_boundary( + Uuid::parse_str("eeeeeeee-2222-4222-8222-eeeeeeee2222").expect("uuid"), + serde_json::json!({ + "accepted_dreaming_review_ref": dreaming_review_ref(accepted_proposal_id, "applied"), + }), + ); + let accepted_proposal = service + .work_journal_entry_create(accepted_proposal_request) + .await + .expect("existing applied dreaming review ref should be accepted"); + + assert_eq!( + accepted_proposal.entry.promotion_boundary["authoritative_memory_allowed"], + serde_json::json!(true) + ); + + test_db.cleanup().await.expect("Failed to cleanup test database."); +} + +async fn insert_active_memory_note(service: &ElfService, note_id: Uuid) { + let now = OffsetDateTime::now_utc(); + + sqlx::query( + "\ +INSERT INTO memory_notes ( + note_id, + tenant_id, + project_id, + agent_id, + scope, + type, + key, + text, + importance, + confidence, + status, + created_at, + updated_at, + expires_at, + embedding_version, + source_ref +) +VALUES ($1,'tenant','project','agent-a','agent_private','fact','accepted-memory-ref','Fact: The accepted memory note is active and readable.',0.8,0.9,'active',$2,$2,NULL,'test:embedding:4096',$3)", + ) + .bind(note_id) + .bind(now) + .bind(serde_json::json!({ "schema": "work_journal_test/v1", "kind": "accepted_memory" })) + .execute(&service.db.pool) + .await + .expect("accepted memory note should insert"); +} + +async fn insert_applied_dreaming_proposal(service: &ElfService, proposal_id: Uuid) { + let run_id = Uuid::parse_str("eeeeeeee-3333-4333-8333-eeeeeeee3333").expect("uuid"); + let now = OffsetDateTime::now_utc(); + + sqlx::query( + "\ +INSERT INTO consolidation_runs ( + run_id, + tenant_id, + project_id, + agent_id, + contract_schema, + job_kind, + status, + input_refs, + source_snapshot, + lineage, + error, + created_at, + updated_at, + completed_at +) +VALUES ($1,'tenant','project','agent-a','elf.consolidation/v1','manual','completed',$2,$3,$4,'{}'::jsonb,$5,$5,$5)", + ) + .bind(run_id) + .bind(serde_json::json!([])) + .bind(serde_json::json!({ "source_count": 0 })) + .bind(serde_json::json!({ "source": "work_journal_test" })) + .bind(now) + .execute(&service.db.pool) + .await + .expect("consolidation run should insert"); + sqlx::query( + "\ +INSERT INTO consolidation_proposals ( + proposal_id, + run_id, + tenant_id, + project_id, + agent_id, + contract_schema, + proposal_kind, + apply_intent, + review_state, + source_refs, + source_snapshot, + lineage, + diff, + confidence, + unsupported_claim_flags, + contradiction_markers, + staleness_markers, + target_ref, + proposed_payload, + reviewer_agent_id, + review_comment, + reviewed_at, + created_at, + updated_at +) +VALUES ($1,$2,'tenant','project','agent-a','elf.consolidation/v1','memory_summary','no_op','applied',$3,$4,$5,$6,0.9,'[]'::jsonb,'[]'::jsonb,'[]'::jsonb,'{}'::jsonb,$7,'agent-a','Apply reviewed Work Journal test proposal.',$8,$8,$8)", + ) + .bind(proposal_id) + .bind(run_id) + .bind(serde_json::json!([])) + .bind(serde_json::json!({ "source_count": 0 })) + .bind(serde_json::json!({ "source": "work_journal_test" })) + .bind(serde_json::json!({ "summary": "Applied proposal supports Work Journal authority." })) + .bind(serde_json::json!({ "schema": "elf.dreaming_review_queue/v1" })) + .bind(now) + .execute(&service.db.pool) + .await + .expect("consolidation proposal should insert"); +} diff --git a/packages/elf-storage/src/lib.rs b/packages/elf-storage/src/lib.rs index 0631dabc..2b377b49 100644 --- a/packages/elf-storage/src/lib.rs +++ b/packages/elf-storage/src/lib.rs @@ -13,6 +13,7 @@ pub mod outbox; pub mod qdrant; pub mod queries; pub mod schema; +pub mod work_journal; mod error; diff --git a/packages/elf-storage/src/models.rs b/packages/elf-storage/src/models.rs index 897ba6fc..2eeda18f 100644 --- a/packages/elf-storage/src/models.rs +++ b/packages/elf-storage/src/models.rs @@ -259,6 +259,47 @@ pub struct GraphPredicateAlias { pub created_at: OffsetDateTime, } +/// Persisted source-adjacent Work Journal entry. +#[derive(Debug, FromRow)] +pub struct WorkJournalEntry { + /// Journal entry identifier. + pub entry_id: Uuid, + /// Tenant that owns the entry. + pub tenant_id: String, + /// Project that owns the entry. + pub project_id: String, + /// Agent that captured the entry. + pub agent_id: String, + /// Visibility scope for readback. + pub scope: String, + /// Stable external or session-local journal session identifier. + pub session_id: String, + /// Entry family discriminator. + pub family: String, + /// Lifecycle status for the journal entry. + pub status: String, + /// Optional display title. + pub title: Option, + /// Redacted durable journal body. + pub body: String, + /// Source references supporting this journal entry. + pub source_refs: Value, + /// Explicit next steps captured from the source. + pub explicit_next_steps: Value, + /// Inferred next steps captured as non-authoritative hints. + pub inferred_next_steps: Value, + /// Options rejected during the captured work session. + pub rejected_options: Value, + /// Promotion boundary metadata for Memory Authority and Dreaming Review. + pub promotion_boundary: Value, + /// Redaction audit for durable journal text. + pub redaction_audit: Value, + /// Creation timestamp. + pub created_at: OffsetDateTime, + /// Last update timestamp. + pub updated_at: OffsetDateTime, +} + /// Persisted supersession row linking two facts. #[derive(Debug, FromRow)] pub struct GraphFactSupersession { diff --git a/packages/elf-storage/src/schema.rs b/packages/elf-storage/src/schema.rs index 9bbafc56..f6d31aef 100644 --- a/packages/elf-storage/src/schema.rs +++ b/packages/elf-storage/src/schema.rs @@ -101,6 +101,8 @@ fn expand_includes(sql: &str) -> String { )), "tables/041_core_memory_block_events.sql" => out .push_str(include_str!("../../../sql/tables/041_core_memory_block_events.sql")), + "tables/042_work_journal_entries.sql" => + out.push_str(include_str!("../../../sql/tables/042_work_journal_entries.sql")), "tables/023_memory_ingest_decisions.sql" => out .push_str(include_str!("../../../sql/tables/023_memory_ingest_decisions.sql")), "tables/024_memory_space_grants.sql" => @@ -136,5 +138,6 @@ mod tests { assert!(schema.contains("CREATE TABLE IF NOT EXISTS core_memory_blocks")); assert!(schema.contains("CREATE TABLE IF NOT EXISTS core_memory_block_attachments")); assert!(schema.contains("CREATE TABLE IF NOT EXISTS core_memory_block_events")); + assert!(schema.contains("CREATE TABLE IF NOT EXISTS work_journal_entries")); } } diff --git a/packages/elf-storage/src/work_journal.rs b/packages/elf-storage/src/work_journal.rs new file mode 100644 index 00000000..71e9aff1 --- /dev/null +++ b/packages/elf-storage/src/work_journal.rs @@ -0,0 +1,158 @@ +//! Work Journal persistence queries. + +use sqlx::PgExecutor; +use uuid::Uuid; + +use crate::{Error, Result, models::WorkJournalEntry}; + +/// Inserts one Work Journal entry row. +pub async fn insert_work_journal_entry<'e, E>(executor: E, entry: &WorkJournalEntry) -> Result<()> +where + E: PgExecutor<'e>, +{ + let result = sqlx::query( + "\ +INSERT INTO work_journal_entries ( + entry_id, + tenant_id, + project_id, + agent_id, + scope, + session_id, + family, + status, + title, + body, + source_refs, + explicit_next_steps, + inferred_next_steps, + rejected_options, + promotion_boundary, + redaction_audit, + created_at, + updated_at +) +VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18) +ON CONFLICT (entry_id) DO NOTHING", + ) + .bind(entry.entry_id) + .bind(entry.tenant_id.as_str()) + .bind(entry.project_id.as_str()) + .bind(entry.agent_id.as_str()) + .bind(entry.scope.as_str()) + .bind(entry.session_id.as_str()) + .bind(entry.family.as_str()) + .bind(entry.status.as_str()) + .bind(entry.title.as_deref()) + .bind(entry.body.as_str()) + .bind(&entry.source_refs) + .bind(&entry.explicit_next_steps) + .bind(&entry.inferred_next_steps) + .bind(&entry.rejected_options) + .bind(&entry.promotion_boundary) + .bind(&entry.redaction_audit) + .bind(entry.created_at) + .bind(entry.updated_at) + .execute(executor) + .await?; + + if result.rows_affected() == 0 { + return Err(Error::Conflict("work_journal entry_id already exists".to_string())); + } + + Ok(()) +} + +/// Fetches one Work Journal entry by tenant and entry identifier. +pub async fn get_work_journal_entry<'e, E>( + executor: E, + tenant_id: &str, + entry_id: Uuid, +) -> Result> +where + E: PgExecutor<'e>, +{ + let row = sqlx::query_as::<_, WorkJournalEntry>( + "\ +SELECT + entry_id, + tenant_id, + project_id, + agent_id, + scope, + session_id, + family, + status, + title, + body, + COALESCE(source_refs, '[]'::jsonb) AS source_refs, + COALESCE(explicit_next_steps, '[]'::jsonb) AS explicit_next_steps, + COALESCE(inferred_next_steps, '[]'::jsonb) AS inferred_next_steps, + COALESCE(rejected_options, '[]'::jsonb) AS rejected_options, + COALESCE(promotion_boundary, '{}'::jsonb) AS promotion_boundary, + COALESCE(redaction_audit, '{}'::jsonb) AS redaction_audit, + created_at, + updated_at +FROM work_journal_entries +WHERE tenant_id = $1 AND entry_id = $2 +LIMIT 1", + ) + .bind(tenant_id) + .bind(entry_id) + .fetch_optional(executor) + .await?; + + Ok(row) +} + +/// Lists recent Work Journal entries for one session in newest-first order. +pub async fn list_work_journal_entries_for_session<'e, E>( + executor: E, + tenant_id: &str, + project_id: &str, + org_project_id: &str, + session_id: &str, + max_rows: i64, +) -> Result> +where + E: PgExecutor<'e>, +{ + let rows = sqlx::query_as::<_, WorkJournalEntry>( + "\ +SELECT + entry_id, + tenant_id, + project_id, + agent_id, + scope, + session_id, + family, + status, + title, + body, + COALESCE(source_refs, '[]'::jsonb) AS source_refs, + COALESCE(explicit_next_steps, '[]'::jsonb) AS explicit_next_steps, + COALESCE(inferred_next_steps, '[]'::jsonb) AS inferred_next_steps, + COALESCE(rejected_options, '[]'::jsonb) AS rejected_options, + COALESCE(promotion_boundary, '{}'::jsonb) AS promotion_boundary, + COALESCE(redaction_audit, '{}'::jsonb) AS redaction_audit, + created_at, + updated_at +FROM work_journal_entries +WHERE tenant_id = $1 + AND project_id IN ($2, $3) + AND session_id = $4 + AND status = 'active' +ORDER BY created_at DESC, entry_id DESC +LIMIT $5", + ) + .bind(tenant_id) + .bind(project_id) + .bind(org_project_id) + .bind(session_id) + .bind(max_rows) + .fetch_all(executor) + .await?; + + Ok(rows) +} diff --git a/sql/init.sql b/sql/init.sql index 99641a31..37333fb1 100644 --- a/sql/init.sql +++ b/sql/init.sql @@ -40,3 +40,4 @@ \ir tables/039_core_memory_blocks.sql \ir tables/040_core_memory_block_attachments.sql \ir tables/041_core_memory_block_events.sql +\ir tables/042_work_journal_entries.sql diff --git a/sql/tables/042_work_journal_entries.sql b/sql/tables/042_work_journal_entries.sql new file mode 100644 index 00000000..cf96f06a --- /dev/null +++ b/sql/tables/042_work_journal_entries.sql @@ -0,0 +1,89 @@ +CREATE TABLE IF NOT EXISTS work_journal_entries ( + entry_id uuid PRIMARY KEY, + tenant_id text NOT NULL, + project_id text NOT NULL, + agent_id text NOT NULL, + scope text NOT NULL, + session_id text NOT NULL, + family text NOT NULL, + status text NOT NULL, + title text NULL, + body text NOT NULL, + source_refs jsonb NOT NULL DEFAULT '[]'::jsonb, + explicit_next_steps jsonb NOT NULL DEFAULT '[]'::jsonb, + inferred_next_steps jsonb NOT NULL DEFAULT '[]'::jsonb, + rejected_options jsonb NOT NULL DEFAULT '[]'::jsonb, + promotion_boundary jsonb NOT NULL DEFAULT '{}'::jsonb, + redaction_audit jsonb NOT NULL DEFAULT '{}'::jsonb, + created_at timestamptz NOT NULL, + updated_at timestamptz NOT NULL +); + +ALTER TABLE work_journal_entries + DROP CONSTRAINT IF EXISTS ck_work_journal_entries_scope; +ALTER TABLE work_journal_entries + ADD CONSTRAINT ck_work_journal_entries_scope + CHECK (scope IN ('agent_private', 'project_shared', 'org_shared')); + +ALTER TABLE work_journal_entries + DROP CONSTRAINT IF EXISTS ck_work_journal_entries_family; +ALTER TABLE work_journal_entries + ADD CONSTRAINT ck_work_journal_entries_family + CHECK ( + family IN ( + 'session_log', + 'handoff_brief', + 'janitor_report', + 'explicit_next_step', + 'inferred_next_step', + 'rejected_option' + ) + ); + +ALTER TABLE work_journal_entries + DROP CONSTRAINT IF EXISTS ck_work_journal_entries_status; +ALTER TABLE work_journal_entries + ADD CONSTRAINT ck_work_journal_entries_status + CHECK (status IN ('active')); + +ALTER TABLE work_journal_entries + DROP CONSTRAINT IF EXISTS ck_work_journal_entries_source_refs; +ALTER TABLE work_journal_entries + ADD CONSTRAINT ck_work_journal_entries_source_refs + CHECK (jsonb_typeof(source_refs) = 'array' AND jsonb_array_length(source_refs) > 0); + +ALTER TABLE work_journal_entries + DROP CONSTRAINT IF EXISTS ck_work_journal_entries_explicit_next_steps; +ALTER TABLE work_journal_entries + ADD CONSTRAINT ck_work_journal_entries_explicit_next_steps + CHECK (jsonb_typeof(explicit_next_steps) = 'array'); + +ALTER TABLE work_journal_entries + DROP CONSTRAINT IF EXISTS ck_work_journal_entries_inferred_next_steps; +ALTER TABLE work_journal_entries + ADD CONSTRAINT ck_work_journal_entries_inferred_next_steps + CHECK (jsonb_typeof(inferred_next_steps) = 'array'); + +ALTER TABLE work_journal_entries + DROP CONSTRAINT IF EXISTS ck_work_journal_entries_rejected_options; +ALTER TABLE work_journal_entries + ADD CONSTRAINT ck_work_journal_entries_rejected_options + CHECK (jsonb_typeof(rejected_options) = 'array'); + +ALTER TABLE work_journal_entries + DROP CONSTRAINT IF EXISTS ck_work_journal_entries_promotion_boundary; +ALTER TABLE work_journal_entries + ADD CONSTRAINT ck_work_journal_entries_promotion_boundary + CHECK (jsonb_typeof(promotion_boundary) = 'object'); + +ALTER TABLE work_journal_entries + DROP CONSTRAINT IF EXISTS ck_work_journal_entries_redaction_audit; +ALTER TABLE work_journal_entries + ADD CONSTRAINT ck_work_journal_entries_redaction_audit + CHECK (jsonb_typeof(redaction_audit) = 'object'); + +CREATE INDEX IF NOT EXISTS idx_work_journal_entries_session + ON work_journal_entries (tenant_id, project_id, session_id, status, created_at DESC); + +CREATE INDEX IF NOT EXISTS idx_work_journal_entries_scope_status + ON work_journal_entries (tenant_id, project_id, scope, status);