fix(streams): enforce repository prompt-sharing policy

README.md

@@ -66,6 +66,10 @@
 
 Installation lets you choose **local-only** mode or **connect to the Autter platform**, and the choice is reversible at any time with `autter onboard`. Local-only attribution, blame, and stats run entirely on your machine — no code, prompts, or usage data ever leave it.
 
+On connected machines, repository prompt-sharing rules also apply to streamed session content. For example, `autter config --add exclude_prompts_in_repositories "https://github.com/acme/private-*"` keeps matching repositories' session content out of the upload queue. When repository restrictions are configured, streams without enough repository context to apply those rules are not uploaded.
+
+After upgrading, session and trace events queued by older versions are held once for consent, since those rows do not record whether repository sharing was permitted when captured. Inspect them with `autter sync status`; use `autter sync import --yes` only if you intend to upload that history, or `autter sync purge --force` to discard the upload backlog.
+
 For verification steps, privacy controls, everyday commands, and troubleshooting, read the **[full installation and setup guide](INSTALL.md)**.
 
 ## Connect to the Autter platform (optional)

src/config.rs

@@ -607,8 +607,18 @@ impl Config {
     /// - User B: autter only in work repos (via allow_repositories), CAS there
     ///   (prompt_storage="default", no include list needed)
     pub fn effective_prompt_storage(&self, repository: &Option<Repository>) -> PromptStorageMode {
+        let remotes = repository
+            .as_ref()
+            .and_then(|repo| repo.remotes_with_urls().ok());
+        self.effective_prompt_storage_with_remotes(remotes.as_ref())
+    }
+
+    fn effective_prompt_storage_with_remotes(
+        &self,
+        remotes: Option<&Vec<(String, String)>>,
+    ) -> PromptStorageMode {
         // Step 1: Check exclusion list first (deny always wins)
-        if self.should_exclude_prompts(repository) {
+        if self.should_exclude_prompts_with_remotes(remotes) {
             return PromptStorageMode::Local;
         }
 
@@ -621,11 +631,7 @@ impl Config {
         }
 
         // Step 3: Check if repo matches include list
-        let remotes = repository
-            .as_ref()
-            .and_then(|repo| repo.remotes_with_urls().ok());
-
-        let matches_include = match &remotes {
+        let matches_include = match remotes {
             Some(remotes) if !remotes.is_empty() => {
                 // Has remotes - check if any match inclusion patterns
                 remotes.iter().any(|remote| {
@@ -654,6 +660,26 @@ impl Config {
         }
     }
 
+    /// Session streams contain conversation content, so both capture and prompt
+    /// sharing rules apply. Unknown repository context cannot satisfy restrictions.
+    pub(crate) fn should_upload_transcript_with_remotes(
+        &self,
+        remotes: Option<&Vec<(String, String)>>,
+    ) -> bool {
+        if !self.platform_sync_enabled() {
+            return false;
+        }
+        if remotes.is_none()
+            && (self.has_repository_filters()
+                || !self.exclude_prompts_in_repositories.is_empty()
+                || !self.include_prompts_in_repositories.is_empty())
+        {
+            return false;
+        }
+        self.is_allowed_repository_with_remotes(remotes)
+            && self.effective_prompt_storage_with_remotes(remotes) != PromptStorageMode::Local
+    }
+
     /// Returns the API key if configured
     pub fn api_key(&self) -> Option<&str> {
         self.api_key.as_deref()
@@ -2390,6 +2416,62 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_transcript_upload_checks_all_repository_remotes() {
+        let config = create_test_config_with_include_prompts(
+            vec![],
+            vec!["https://github.com/private/*".into()],
+            "default",
+            None,
+        );
+        let public = vec![("origin".into(), "https://github.com/public/repo".into())];
+        assert!(config.should_upload_transcript_with_remotes(Some(&public)));
+        let mut multiple = public;
+        multiple.push(("mirror".into(), "git@github.com:private/repo.git".into()));
+        assert!(!config.should_upload_transcript_with_remotes(Some(&multiple)));
+        assert!(!config.should_upload_transcript_with_remotes(None));
+    }
+
+    #[test]
+    fn test_transcript_upload_honors_include_fallback_and_local_mode() {
+        let work = vec![("origin".into(), "https://github.com/work/repo".into())];
+        let other = vec![("origin".into(), "https://github.com/personal/repo".into())];
+        let mut config = create_test_config_with_include_prompts(
+            vec!["https://github.com/work/*".into()],
+            vec![],
+            "default",
+            Some("local"),
+        );
+        assert!(config.should_upload_transcript_with_remotes(Some(&work)));
+        assert!(!config.should_upload_transcript_with_remotes(Some(&other)));
+        assert!(!config.should_upload_transcript_with_remotes(None));
+        config.prompt_storage = "local".into();
+        config.default_prompt_storage = Some("default".into());
+        assert!(!config.should_upload_transcript_with_remotes(Some(&other)));
+    }
+
+    #[test]
+    fn test_transcript_upload_honors_capture_filters() {
+        let allowed = vec![("origin".into(), "https://github.com/allowed/repo".into())];
+        let private = vec![("origin".into(), "https://github.com/private/repo".into())];
+        let config = create_test_config(vec!["https://github.com/allowed/*".into()], vec![]);
+        assert!(config.should_upload_transcript_with_remotes(Some(&allowed)));
+        assert!(!config.should_upload_transcript_with_remotes(Some(&private)));
+        assert!(!config.should_upload_transcript_with_remotes(None));
+        let config = create_test_config(vec![], vec!["https://github.com/private/*".into()]);
+        assert!(!config.should_upload_transcript_with_remotes(Some(&private)));
+    }
+
+    #[test]
+    fn test_transcript_upload_unknown_context_without_restrictions() {
+        let config = create_test_config(vec![], vec![]);
+        assert!(config.should_upload_transcript_with_remotes(None));
+        let excluded =
+            create_test_config_with_include_prompts(vec![], vec!["*".into()], "default", None);
+        assert!(!excluded.should_upload_transcript_with_remotes(Some(&vec![])));
+        assert!(!excluded.should_upload_transcript_with_remotes(None));
+    }
+
     #[test]
     fn test_update_channel_default_is_latest() {
         let channel = UpdateChannel::default();

src/daemon/stream_worker.rs

@@ -944,6 +944,14 @@
             base_attrs = base_attrs.repo_url(url);
         }
 
+        // Use every remote, matching the repository policy used by commit-time
+        // transcript capture. A shared stream has no single repository context.
+        let remotes = resolved_work_dir
+            .as_deref()
+            .and_then(|cwd| {
+                crate::git::repository::discover_repository_in_path_no_git_exec(cwd).ok()
+            })
+            .and_then(|repo| repo.remotes_with_urls().ok());
         let file_meta = std::fs::metadata(&path).ok();
         let watermark_type_str = &stream.watermark_type;
         let is_initial_watermark = stream.watermark_value.is_empty()
@@ -972,10 +980,17 @@
 
             let batch_count = batch.events.len();
 
+            // Re-read privacy settings for each batch in this long-lived daemon.
+            // Still advance the cursor for withheld content so enabling sharing
+            // later does not silently replay previously private conversations.
+            let upload_allowed =
+                config::Config::fresh().should_upload_transcript_with_remotes(remotes.as_ref());
+
             let is_otel_stream = task.stream_kind == "otel_traces";
             let metric_events: Vec<MetricEvent> = batch
                 .events
                 .into_iter()
+                .filter(|_| upload_allowed)
                 .enumerate()
                 .filter_map(|(idx, raw_event)| {
                     let (eid, pid, tid) = agent.extract_event_ids(&raw_event);
@@ -1025,7 +1040,13 @@
                 })
                 .collect();
 
-            crate::observability::log_metrics(metric_events);
+            if !metric_events.is_empty() {
+                telemetry.submit_telemetry_blocking(vec![
+                    crate::daemon::TelemetryEnvelope::Metrics {
+                        events: metric_events,
+                    },
+                ]);
+            }
 
             // Backpressure: if the telemetry buffer has accumulated too many
             // events, poll briefly to let the 3-second flush cycle drain it.

src/daemon/telemetry_worker.rs

@@ -141,6 +141,12 @@ impl DaemonTelemetryWorkerHandle {
         self.buffer.lock().await.ingest_envelopes(envelopes);
     }
 
+    /// Submit from a blocking worker without routing through global diagnostic
+    /// logging, which is best-effort and disabled in test-support builds.
+    pub fn submit_telemetry_blocking(&self, envelopes: Vec<TelemetryEnvelope>) {
+        self.buffer.blocking_lock().ingest_envelopes(envelopes);
+    }
+
     /// Submit CAS records for batched upload.
     pub async fn submit_cas(&self, records: Vec<CasSyncPayload>) {
         self.buffer.lock().await.ingest_cas(records);

src/metrics/db.rs

@@ -315,6 +315,30 @@ impl MetricsDatabase {
         Ok(n)
     }
 
+    /// Older transcript rows lack the capture-time sharing decision. Preserve
+    /// them locally until explicit import, leaving ordinary metrics eligible.
+    pub fn hold_transcript_events(&mut self) -> Result<usize, AutterError> {
+        use crate::metrics::types::MetricEventId;
+        let tx = self.conn.transaction()?;
+        let session_kind = MetricEventId::SessionEvent as u16;
+        let trace_kind = MetricEventId::OtelTrace as u16;
+        let kinds = params![session_kind, trace_kind];
+        let n = tx.execute(
+            "INSERT INTO metrics_held (event_json)
+             SELECT event_json FROM metrics
+             WHERE CASE WHEN json_valid(event_json) THEN json_extract(event_json, '$.e') END IN (?1, ?2)
+             ORDER BY id",
+            kinds,
+        )?;
+        tx.execute(
+            "DELETE FROM metrics
+             WHERE CASE WHEN json_valid(event_json) THEN json_extract(event_json, '$.e') END IN (?1, ?2)",
+            kinds,
+        )?;
+        tx.commit()?;
+        Ok(n)
+    }
+
     /// Move held events back into the upload queue.
     pub fn release_held(&mut self) -> Result<usize, AutterError> {
         let tx = self.conn.transaction()?;
@@ -421,6 +445,29 @@ mod tests {
         assert_eq!(db.delete_all().unwrap(), 3, "purge removes held events too");
     }
 
+    #[test]
+    fn test_transcript_hold_preserves_content_and_other_metrics() {
+        let (mut db, _temp_dir) = create_test_db();
+        let events = vec![
+            r#"{"e":1,"v":{"0":"ordinary metric"}}"#.to_string(),
+            r#"{"e":5,"v":{"0":"private conversation"}}"#.to_string(),
+            r#"{"e":6,"v":{"0":"private trace"}}"#.to_string(),
+            "malformed old record".to_string(),
+        ];
+        db.insert_events(&events).unwrap();
+        assert_eq!(db.hold_transcript_events().unwrap(), 2);
+        assert_eq!(db.count_held().unwrap(), 2);
+        let pending = db.get_batch(10).unwrap();
+        assert_eq!(pending.len(), 2);
+        assert_eq!(pending[0].event_json, events[0]);
+        assert_eq!(pending[1].event_json, events[3]);
+        assert_eq!(db.hold_transcript_events().unwrap(), 0);
+        assert_eq!(db.release_held().unwrap(), 2);
+        let restored = db.get_batch(10).unwrap();
+        assert_eq!(restored[2].event_json, events[1]);
+        assert_eq!(restored[3].event_json, events[2]);
+    }
+
     #[test]
     fn test_initialize_schema() {
         let (db, _temp_dir) = create_test_db();

src/upload_hold.rs

@@ -16,6 +16,10 @@
 //!   current backlog before signing in, so the background service cannot drain
 //!   it while onboarding asks what to do with it.
 //!
+//! A separate one-time hold covers session/trace metrics queued before the
+//! stream worker enforced repository sharing rules. The first migration's
+//! marker alone cannot establish consent for those subsequently queued rows.
+//!
 //! Provenance going forward needs no new columns: since local mode stopped
 //! queueing upload data and excluded repos stopped queueing transcripts, every
 //! item queued after the migration was queued in connected mode with sharing
@@ -40,12 +44,17 @@ pub const HELD_RETRY_AT: i64 = i64::MAX;
 
 /// Bump to re-run [`ensure_legacy_migration`] on every machine.
 pub const LEGACY_MIGRATION_VERSION: u32 = 1;
+const TRANSCRIPT_PRIVACY_MIGRATION_VERSION: u32 = 1;
 
 #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
 struct HoldState {
     /// Last [`LEGACY_MIGRATION_VERSION`] applied on this machine.
     #[serde(default)]
     legacy_migration: u32,
+    /// Session content could still bypass repository sharing after the first
+    /// migration. This hold affects only transcript/trace metrics, not all queues.
+    #[serde(default)]
+    transcript_privacy_migration: u32,
 }
 
 /// Per-queue item counts.
@@ -184,17 +193,35 @@ pub fn held_counts() -> QueueCounts {
 /// `Err` when it could not be applied, in which case callers must not drain.
 pub fn ensure_legacy_migration() -> Result<bool, String> {
     let mut state = load_state();
-    if state.legacy_migration >= LEGACY_MIGRATION_VERSION {
+    let legacy_pending = state.legacy_migration < LEGACY_MIGRATION_VERSION;
+    let transcripts_pending =
+        state.transcript_privacy_migration < TRANSCRIPT_PRIVACY_MIGRATION_VERSION;
+    if !legacy_pending && !transcripts_pending {
         return Ok(true);
     }
-    let held = hold_all()?;
-    if held.total() > 0 {
-        tracing::info!(
-            held = %held.summary(),
-            "upload hold: holding backlog queued by an earlier autter version until the user consents"
-        );
+    if legacy_pending {
+        let held = hold_all()?;
+        if held.total() > 0 {
+            tracing::info!(
+                held = %held.summary(),
+                "upload hold: holding backlog queued by an earlier autter version until the user consents"
+            );
+        }
+        state.legacy_migration = LEGACY_MIGRATION_VERSION;
+    }
+    if transcripts_pending {
+        let db = crate::metrics::db::MetricsDatabase::global().map_err(|e| e.to_string())?;
+        let held = lock(db)?
+            .hold_transcript_events()
+            .map_err(|e| e.to_string())?;
+        if held > 0 {
+            tracing::info!(
+                held,
+                "upload hold: holding older session content until the user consents"
+            );
+        }
+        state.transcript_privacy_migration = TRANSCRIPT_PRIVACY_MIGRATION_VERSION;
     }
-    state.legacy_migration = LEGACY_MIGRATION_VERSION;
     save_state(&state)?;
     Ok(true)
 }
@@ -215,8 +242,10 @@ mod tests {
     fn hold_state_round_trips_and_defaults_to_unmigrated() {
         let state: HoldState = serde_json::from_str("{}").unwrap();
         assert_eq!(state.legacy_migration, 0);
+        assert_eq!(state.transcript_privacy_migration, 0);
         let state = HoldState {
             legacy_migration: LEGACY_MIGRATION_VERSION,
+            transcript_privacy_migration: TRANSCRIPT_PRIVACY_MIGRATION_VERSION,
         };
         let back: HoldState = serde_json::from_slice(&serde_json::to_vec(&state).unwrap()).unwrap();
         assert_eq!(back, state);