feat: implement I/O-aware RAM fallback router and native string-indexed memory

c/coli

@@ -159,8 +159,27 @@ def resource_request(a, env):
         [int(value) for value in gpu.split(",")])
     return ram,ctx,devices,vram
 
+def detect_slow_drive(model_path):
+    if os.environ.get("COLI_FORCE_NVME") == "1": return False
+    if os.environ.get("COLI_FORCE_HDD") == "1": return True
+    try:
+        if sys.platform == "win32":
+            out = subprocess.check_output(["powershell", "-NoProfile", "-Command", "Get-PhysicalDisk | Select-Object -ExpandProperty MediaType"], text=True, timeout=2).strip().lower()
+            return "hdd" in out
+        elif sys.platform.startswith("linux"):
+            out = subprocess.check_output(["lsblk", "-ndo", "ROTA", model_path], text=True, timeout=2).strip()
+            return out == "1"
+    except Exception:
+        pass
+    return False
+
 def env_for(a):
     e = dict(os.environ, SNAP=a.model)
+    if detect_slow_drive(a.model):
+        print(f"  {C.org}[AUTO-DETECT] Mechanical HDD detected. Enabling RAM-Aware Fallback Routing.{C.r}", file=sys.stderr)
+        e["COLI_HDD_MODE"] = "1"
+        e["AUTO_LATENCY_TARGET_MS"] = "500"
+
     e["COLI_POLICY"]=a.policy
     if a.ram:  e["RAM_GB"]=str(a.ram)
     if a.ngen: e["NGEN"]=str(a.ngen)

c/glm.c

@@ -843,6 +843,10 @@ static float g_temp=-1;  /* TEMP: temperatura di sampling sui TOKEN. <0 = auto (
 static float g_nuc=0.95f;/* NUCLEUS: top-p sul vocabolario (default dal generation_config GLM-5.2) */
 static int g_topk=0;     /* TOPK=n -> usa n expert/token invece di config (ricerca: meno disco) */
 static float g_topp=0;   /* TOPP=p (0..1) -> top-p adattivo: tieni gli expert fino a peso cumulato p */
+static int g_hdd_mode=0;
+static double g_auto_latency_target=0.0;
+static double g_ewait_latency=0.0;
+static uint64_t g_ewait_samples=0;
 /* CACHE_ROUTE (paper 2412.00099 max-rank): opt-in only. Keep true top-J always;
  * fill remaining slots preferring pin∪LRU experts ranked within top-M (or mass ROUTE_P). */
 static int g_cache_route=0;
@@ -2016,7 +2020,7 @@ static void attention_rows(Model *m, Layer *l, int layer, float *x, int S, int p
         for(int s=0;s<S;s++) for(int h=0;h<H;h++){
             KVState *ks=kvs?kvs[s]:m->kv;
             int pos=positions?positions[s]:pos_base+s;
-            const float *qp=Q+(int64_t)s*H*qh+(int64_t)h*qh;
+            const float *qp=Q+(int64_t)s*H*qh+(int64_t)h*qh;          /* [qk_nope | qk_rope] */
             const float *qr=qp+c->qk_nope;
             int rbase=h*(c->qk_nope+vh);
             float qabs[512]; memset(qabs,0,kvl*sizeof(float));
@@ -2200,6 +2204,11 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
                     idx[chosen]=e; w[chosen]=rank_w[r]; chosen++;
                 }
             }
+            if(g_hdd_mode && g_auto_latency_target > 0 && g_ewait_latency > g_auto_latency_target && chosen > 0) {
+                /* RAM-Aware Fallback: disk is too slow, STOP looking for uncached experts */
+                Ksel = chosen; 
+            }
+
             /* Fill remainder from true ranking order. */
             for(int r=0;r<Mwin && chosen<Ksel;r++){
                 int e=rank_buf[r]; int already=0;
@@ -2249,9 +2258,19 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
         } else {
             for(int kk=0;kk<Ksel;kk++){ int best=-1; float bv=-1e30f;
                 for(int e=0;e<E;e++){ int tk=0; for(int j=0;j<kk;j++) if(idx[j]==e){tk=1;break;}
-                    if(!tk && choice[e]>bv){bv=choice[e];best=e;} }
+                    int eligible = 1;
+                    if(g_hdd_mode && g_auto_latency_target > 0 && g_ewait_latency > g_auto_latency_target && kk > 0) {
+                        eligible = expert_is_resident(m,layer,e);
+                    }
+                    if(!tk && eligible && choice[e]>bv){bv=choice[e];best=e;} }
+                if(best<0) break;
                 idx[kk]=best; w[kk]=logit[best];
             }
+            if(g_hdd_mode && g_auto_latency_target > 0 && g_ewait_latency > g_auto_latency_target) {
+                int c_k = 0; for(int kk=0;kk<Ksel;kk++) if(idx[kk]>=0 && w[kk] > -1e20f) c_k++;
+                Ksel = c_k > 0 ? c_k : 1; /* Fallback to 1 if nothing */
+            }
+
             if(g_route_agree){
                 m->route_agree_hit+=(uint64_t)Ksel;
                 m->route_agree_tot+=(uint64_t)Ksel;
@@ -2398,10 +2417,21 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
                 int eids[64]; for(int q=0;q<nmiss;q++) eids[q]=uniq[base+missk[q]];
                 pipe_dispatch(m,layer,eids,nmiss);
                 m->t_edisk += now_s()-t0;           /* dispatch only; real reads hide behind matmul */
-            } else { double t0=now_s();             /* ORIGINALE: blocking parallel load */
-                #pragma omp parallel for schedule(dynamic,1)
-                for(int q=0;q<nmiss;q++) expert_load(m,layer,uniq[base+missk[q]],&m->ws[q],1);
-                m->t_edisk += now_s()-t0; }
+            } else {
+                double t0=now_s();
+                for(int q=0;q<nmiss;q++){
+                    int e=uniq[base+missk[q]]; ESlot *s=use[missk[q]];
+                    expert_load(m,layer,e,s,0);
+                }
+                double read_ms = (now_s()-t0) * 1000.0;
+                m->t_edisk += read_ms/1000.0;
+                if(g_hdd_mode && nmiss > 0) {
+                    double per_miss = read_ms / nmiss;
+                    if(g_ewait_samples == 0) g_ewait_latency = per_miss;
+                    else g_ewait_latency = g_ewait_latency * 0.8 + per_miss * 0.2;
+                    g_ewait_samples++;
+                }
+            }
         }
         /* I/O ASINCRONO: readahead (WILLNEED) del blocco SUCCESSIVO mentre calcoliamo
          * questo — il kernel legge in background, le pread dopo trovano cache calda */
@@ -4763,6 +4793,9 @@ int main(int argc, char **argv){
     if(g_mmap) fprintf(stderr,"[MMAP] expert = viste zero-copy nei file (page cache = cache)\n");
     g_topk = getenv("TOPK")?atoi(getenv("TOPK")):0;
     g_topp = getenv("TOPP")?atof(getenv("TOPP")):0;
+    g_hdd_mode = getenv("COLI_HDD_MODE")?atoi(getenv("COLI_HDD_MODE")):0;
+    g_auto_latency_target = getenv("AUTO_LATENCY_TARGET_MS")?atof(getenv("AUTO_LATENCY_TARGET_MS")):0.0;
+
     g_cache_route = getenv("CACHE_ROUTE")?atoi(getenv("CACHE_ROUTE")):0;
     g_route_j = getenv("ROUTE_J")?atoi(getenv("ROUTE_J")):2;
     g_route_m = getenv("ROUTE_M")?atoi(getenv("ROUTE_M")):12;

c/openai_server.py

@@ -58,6 +58,73 @@ def error_object(error):
                       "param": error.param, "code": error.code}}
 
 
+class MemoryManager:
+    def __init__(self):
+        self.base_dir = Path.home() / ".colibri" / "memory"
+        self.base_dir.mkdir(parents=True, exist_ok=True)
+        self.lock = threading.Lock()
+
+    def _get_path(self, user):
+        safe_user = "".join(c for c in str(user) if c.isalnum() or c in "._-")
+        if not safe_user: return None
+        return self.base_dir / f"{safe_user}.txt"
+
+    def search_and_inject(self, user, messages):
+        if not user or not messages: return
+        path = self._get_path(user)
+        if not path or not path.exists(): return
+
+        last_user_msg = ""
+        for m in reversed(messages):
+            if m.get("role") == "user":
+                content = m.get("content", "")
+                if isinstance(content, str): last_user_msg = content
+                elif isinstance(content, list): last_user_msg = " ".join(p.get("text", "") for p in content if p.get("type") == "text")
+                break
+
+        if not last_user_msg: return
+        words = [w.strip(".,!?\"'()[]{}:;") for w in last_user_msg.split()]
+        keywords = {w.lower() for w in words if len(w) > 4}
+        if not keywords: return
+
+        matched_lines = []
+        try:
+            with self.lock:
+                lines = path.read_text(encoding="utf-8").splitlines()
+                for line in lines:
+                    line_lower = line.lower()
+                    if any(kw in line_lower for kw in keywords):
+                        matched_lines.append(line)
+        except Exception:
+            return
+
+        if matched_lines:
+            # deduplicate
+            unique = list(dict.fromkeys(matched_lines))
+            # take top 10 matches
+            unique = unique[-10:]
+            injection = "[LONG-TERM MEMORY]\n" + "\n".join(unique) + "\n[/LONG-TERM MEMORY]\n\n"
+            if messages[0].get("role") == "system":
+                messages[0]["content"] = injection + (messages[0].get("content") or "")
+            else:
+                messages.insert(0, {"role": "system", "content": injection})
+
+    def append_journal(self, user, user_msg, assistant_msg):
+        if not user or not user_msg or not assistant_msg: return
+        path = self._get_path(user)
+        if not path: return
+        try:
+            with self.lock:
+                with open(path, "a", encoding="utf-8") as f:
+                    # Clean up newlines for a more compact PDF-like single-line search index
+                    u_clean = user_msg.replace('\n', ' ').strip()
+                    a_clean = assistant_msg.replace('\n', ' ').strip()
+                    f.write(f"User: {u_clean} | Assistant: {a_clean}\n")
+        except Exception:
+            pass
+
+GLOBAL_MEMORY_MANAGER = MemoryManager()
+
 class GenerationScheduler:
     """Bounded FIFO admission for the engine's independent KV contexts."""
 
@@ -826,7 +893,7 @@ def do_POST(self):
             except OSError:
                 pass
 
-    def generation(self, body, prompt, request_id, chat):
+    def generation(self, body, prompt, request_id, chat, user_id=None, last_user_msg=None):
         # COLI_DEBUG tees the engine transaction to stderr: 1 = decoded output stream only,
         # 2 = both sides (rendered prompt + output). render_chat already folds prior turns and
         # tool results into `prompt`, so level 2 is the full conversation the engine saw.
@@ -880,7 +947,12 @@ def generation(self, body, prompt, request_id, chat):
                     choice = ({"index": 0, "message": {"role": "assistant", "content": text,
                                "refusal": None}, "logprobs": None, "finish_reason": length_finish} if chat else
                               {"index": 0, "text": text, "logprobs": None, "finish_reason": length_finish})
+
+                if user_id and last_user_msg:
+                    GLOBAL_MEMORY_MANAGER.append_journal(user_id, last_user_msg, text)
+
                 self.send_json(200, {"id": completion_id, "object": object_name, "created": created,
+
                     "model": self.server.model_id, "choices": [choice], "usage": self.usage(stats)},
                     request_id, queue_headers)
                 return
@@ -976,22 +1048,32 @@ def emit_tools(chunk):
                     lambda: not connected)
                 if not sp["tool"] and sp["buf"]:
                     emit(sp["buf"])                     # no tool call happened: flush held tail
+
+                if user_id and last_user_msg:
+                    GLOBAL_MEMORY_MANAGER.append_journal(user_id, last_user_msg, "".join(raw))
+
                 _content, calls = parse_tool_calls("".join(raw), tools)
+
                 for i, tc in enumerate(calls):
                     event([{"index": 0, "delta": {"tool_calls": [{"index": i, "id": tc["id"],
                              "type": "function", "function": {"name": tc["function"]["name"],
                              "arguments": tc["function"]["arguments"]}}]},
                             "logprobs": None, "finish_reason": None}])
                 finish = "tool_calls" if calls else ("length" if stats["length_limited"] else "stop")
             else:
+                raw_text_acc = []
                 def emit_plain(chunk):
+                    raw_text_acc.append(chunk)
                     if dbg_echo:
                         sys.stderr.write(chunk); sys.stderr.flush()
                     emit(chunk)
                 stats = self.server.engine.generate(
                     prompt, maximum, temperature, top_p, emit_plain, cache_slot,
                     lambda: not connected)
+                if user_id and last_user_msg:
+                    GLOBAL_MEMORY_MANAGER.append_journal(user_id, last_user_msg, "".join(raw_text_acc))
                 finish = "length" if stats["length_limited"] else "stop"
+
             ka_stop.set()                          # generation done: stop the keepalive pump
             ka_thread.join(timeout=2)
             final_choice = ({"index": 0, "delta": {}, "logprobs": None, "finish_reason": finish}
@@ -1000,7 +1082,9 @@ def emit_plain(chunk):
             event([final_choice])
             if include_usage:
                 event([], self.usage(stats))
+
             if connected:
+
                 try:
                     self.wfile.write(b"data: [DONE]\n\n")
                     self.wfile.flush()
@@ -1041,9 +1125,23 @@ def chat_completion(self, body, request_id):
         if not isinstance(enable_thinking, bool):
             raise APIError(400, "`enable_thinking` must be a boolean.", "enable_thinking")
         tools = body.get("tools") or body.get("functions") or None
-        prompt = render_chat(body.get("messages"), enable_thinking, reasoning_effort, tools,
+
+        messages = body.get("messages", [])
+        user_id = body.get("user")
+        last_user_msg = ""
+        if user_id:
+            # Capture the last user message to append to the journal later
+            for m in reversed(messages):
+                if m.get("role") == "user":
+                    c = m.get("content", "")
+                    last_user_msg = c if isinstance(c, str) else " ".join(p.get("text", "") for p in c if p.get("type") == "text")
+                    break
+            GLOBAL_MEMORY_MANAGER.search_and_inject(user_id, messages)
+
+        prompt = render_chat(messages, enable_thinking, reasoning_effort, tools,
                              body.get("tool_choice"))
-        self.generation(body, prompt, request_id, True)
+        self.generation(body, prompt, request_id, True, user_id=user_id, last_user_msg=last_user_msg)
+
 
     def completion(self, body, request_id):
         prompt = body.get("prompt")