observation agent rewrite, edit command, daemon fixes

- observation.agent: rewritten to navigate graph and prefer refining
  existing nodes over creating new ones. Identity-framed prompt,
  goals over rules.
- poc-memory edit: opens node in $EDITOR, writes back on save,
  no-op if unchanged
- daemon: remove extra_workers (jobkit tokio migration dropped it),
  remove sequential chaining of same-type agents (in-flight exclusion
  is sufficient)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

poc-memory/src/agents/daemon.rs

@@ -698,7 +698,6 @@ pub fn run_daemon() -> Result<(), String> {
         data_dir: config.data_dir.clone(),
         resource_slots: config.llm_concurrency,
         resource_name: "llm".to_string(),
-        extra_workers: 3,
     });
 
     let choir = Arc::clone(&daemon.choir);
@@ -1043,30 +1042,26 @@ pub fn run_daemon() -> Result<(), String> {
                 log_event("scheduler", "consolidation-plan",
                     &format!("{} agents ({})", runs.len(), summary.join(" ")));
 
-                // Phase 1: Agent runs — sequential within type, parallel across types.
-                // Same-type agents chain (they may touch overlapping graph regions),
-                // but different types run concurrently (different seed nodes).
-                let mut prev_by_type: std::collections::HashMap<String, jobkit::RunningTask> =
-                    std::collections::HashMap::new();
+                // Phase 1: Agent runs — all concurrent, in-flight exclusion
+                // prevents overlapping graph regions.
+                let mut all_tasks: Vec<jobkit::RunningTask> = Vec::new();
                 for (i, (agent_type, batch)) in runs.iter().enumerate() {
                     let agent = agent_type.to_string();
                     let b = *batch;
                     let in_flight_clone = Arc::clone(&in_flight_sched);
                     let task_name = format!("c-{}-{}:{}", agent, i, today);
-                    let mut builder = choir_sched.spawn(task_name)
+                    let task = choir_sched.spawn(task_name)
                         .resource(&llm_sched)
                         .log_dir(&log_dir_sched)
                         .retries(1)
                         .init(move |ctx| {
                             job_consolidation_agent(ctx, &agent, b, &in_flight_clone)
-                        });
-                    if let Some(dep) = prev_by_type.get(agent_type.as_str()) {
-                        builder.depend_on(dep);
-                    }
-                    prev_by_type.insert(agent_type.clone(), builder.run());
+                        })
+                        .run();
+                    all_tasks.push(task);
                 }
-                // Orphans phase depends on all agent type chains completing
-                let prev_agent = prev_by_type.into_values().last();
+                // Orphans phase depends on all agent tasks completing
+                let prev_agent = all_tasks.last().cloned();
 
                 // Phase 2: Link orphans (CPU-only, no LLM)
                 let mut orphans = choir_sched.spawn(format!("c-orphans:{}", today))

poc-memory/src/cli/node.rs

@@ -399,6 +399,60 @@ pub fn cmd_write(key: &[String]) -> Result<(), String> {
     Ok(())
 }
 
+pub fn cmd_edit(key: &[String]) -> Result<(), String> {
+    if key.is_empty() {
+        return Err("edit requires a key".into());
+    }
+    let raw_key = key.join(" ");
+    let store = store::Store::load()?;
+    let key = store.resolve_key(&raw_key).unwrap_or(raw_key.clone());
+
+    let content = store.nodes.get(&key)
+        .map(|n| n.content.clone())
+        .unwrap_or_default();
+
+    let tmp = std::env::temp_dir().join(format!("poc-memory-edit-{}.md", key.replace('/', "_")));
+    std::fs::write(&tmp, &content)
+        .map_err(|e| format!("write temp file: {}", e))?;
+
+    let editor = std::env::var("EDITOR").unwrap_or_else(|_| "vi".into());
+    let status = std::process::Command::new(&editor)
+        .arg(&tmp)
+        .status()
+        .map_err(|e| format!("spawn {}: {}", editor, e))?;
+
+    if !status.success() {
+        let _ = std::fs::remove_file(&tmp);
+        return Err(format!("{} exited with {}", editor, status));
+    }
+
+    let new_content = std::fs::read_to_string(&tmp)
+        .map_err(|e| format!("read temp file: {}", e))?;
+    let _ = std::fs::remove_file(&tmp);
+
+    if new_content == content {
+        println!("No change: '{}'", key);
+        return Ok(());
+    }
+
+    if new_content.trim().is_empty() {
+        return Err("Content is empty, aborting".into());
+    }
+
+    drop(store);
+    let mut store = store::Store::load()?;
+    let result = store.upsert(&key, &new_content)?;
+    match result {
+        "unchanged" => println!("No change: '{}'", key),
+        "updated" => println!("Updated '{}' (v{})", key, store.nodes[&key].version),
+        _ => println!("Created '{}'", key),
+    }
+    if result != "unchanged" {
+        store.save()?;
+    }
+    Ok(())
+}
+
 pub fn cmd_lookup_bump(keys: &[String]) -> Result<(), String> {
     if keys.is_empty() {
         return Err("lookup-bump requires at least one key".into());

poc-memory/src/main.rs

@@ -69,6 +69,11 @@ enum Command {
         /// Node key
         key: Vec<String>,
     },
+    /// Edit a node in $EDITOR
+    Edit {
+        /// Node key
+        key: Vec<String>,
+    },
     /// Show all stored versions of a node
     History {
         /// Show full content for every version
@@ -778,6 +783,7 @@ fn main() {
             => cli::misc::cmd_search(&query, &pipeline, expand, full, debug, fuzzy, content),
         Command::Render { key } => cli::node::cmd_render(&key),
         Command::Write { key } => cli::node::cmd_write(&key),
+        Command::Edit { key } => cli::node::cmd_edit(&key),
         Command::History { full, key } => cli::node::cmd_history(&key, full),
         Command::Tail { n, full } => cli::journal::cmd_tail(n, full),
         Command::Status => cli::misc::cmd_status(),