fix: close file handles via context manager in remaining call sites

agentmain.py

@@ -13,24 +13,28 @@
 script_dir = os.path.dirname(os.path.abspath(__file__))
 def load_tool_schema(suffix=''):
     global TOOLS_SCHEMA
-    TS = open(os.path.join(script_dir, f'assets/tools_schema{suffix}.json'), 'r', encoding='utf-8').read()
+    with open(os.path.join(script_dir, f'assets/tools_schema{suffix}.json'), 'r', encoding='utf-8') as f: TS = f.read()
     TOOLS_SCHEMA = json.loads(TS if os.name == 'nt' else TS.replace('powershell', 'bash'))
 load_tool_schema()
 
 lang_suffix = '_en' if os.environ.get('GA_LANG', '') == 'en' else ''
 mem_dir = os.path.join(script_dir, 'memory')
 if not os.path.exists(mem_dir): os.makedirs(mem_dir)
 mem_txt = os.path.join(mem_dir, 'global_mem.txt')
-if not os.path.exists(mem_txt): open(mem_txt, 'w', encoding='utf-8').write('# [Global Memory - L2]\n')
+if not os.path.exists(mem_txt):
+    with open(mem_txt, 'w', encoding='utf-8') as f: f.write('# [Global Memory - L2]\n')
 mem_insight = os.path.join(mem_dir, 'global_mem_insight.txt')
 if not os.path.exists(mem_insight):
     t = os.path.join(script_dir, f'assets/global_mem_insight_template{lang_suffix}.txt')
-    open(mem_insight, 'w', encoding='utf-8').write(open(t, encoding='utf-8').read() if os.path.exists(t) else '')
+    if os.path.exists(t):
+        with open(t, encoding='utf-8') as _f: _seed = _f.read()
+    else: _seed = ''
+    with open(mem_insight, 'w', encoding='utf-8') as f: f.write(_seed)
 cdp_cfg = os.path.join(script_dir, 'assets/tmwd_cdp_bridge/config.js')
 if not os.path.exists(cdp_cfg):
     try:
         os.makedirs(os.path.dirname(cdp_cfg), exist_ok=True)
-        open(cdp_cfg, 'w', encoding='utf-8').write(f"const TID = '__ljq_{hex(random.randint(0, 99999999))[2:8]}';")
+        with open(cdp_cfg, 'w', encoding='utf-8') as f: f.write(f"const TID = '__ljq_{hex(random.randint(0, 99999999))[2:8]}';")
     except Exception as e: print(f'[WARN] CDP config init failed: {e} — advanced web features (tmwebdriver) will be unavailable.')
 
 def get_system_prompt():
@@ -102,7 +106,8 @@ def _handle_slash_cmd(self, raw_query, display_queue):
         if _sm := re.match(r'/session\.(\w+)=(.*)', raw_query.strip()):
             k, v = _sm.group(1), _sm.group(2)
             vfile = os.path.join(script_dir, 'temp', v)
-            if os.path.isfile(vfile): v = open(vfile, encoding='utf-8').read().strip()
+            if os.path.isfile(vfile):
+                with open(vfile, encoding='utf-8') as f: v = f.read().strip()
             try: v = json.loads(v)  # cover number parsing
             except (json.JSONDecodeError, ValueError): pass
             setattr(self.llmclient.backend, k, v)
@@ -239,7 +244,7 @@ def run(self):
                 print(f'[Reflect] drain error: {e}'); result = f'[ERROR] {e}'
             log_dir = os.path.join(script_dir, 'temp/reflect_logs'); os.makedirs(log_dir, exist_ok=True)
             script_name = os.path.splitext(os.path.basename(args.reflect))[0]
-            open(os.path.join(log_dir, f'{script_name}_{datetime.now():%Y-%m-%d}.log'), 'a', encoding='utf-8').write(f'[{datetime.now():%m-%d %H:%M}]\n{result}\n\n')
+            with open(os.path.join(log_dir, f'{script_name}_{datetime.now():%Y-%m-%d}.log'), 'a', encoding='utf-8') as f: f.write(f'[{datetime.now():%m-%d %H:%M}]\n{result}\n\n')
             if (on_done := getattr(mod, 'on_done', None)):
                 try: on_done(result)
                 except Exception as e: print(f'[Reflect] on_done error: {e}')

frontends/wechatapp.py

@@ -180,7 +180,8 @@ def _dl_media(items):
                 ct = requests.get(f'{CDN_BASE}/download?encrypted_query_param={quote(eq)}', timeout=60).content
                 pt = AES.new(aes_key, AES.MODE_ECB).decrypt(ct); pt = pt[:-pt[-1]]
                 fname = sub.get('file_name') or f'{uuid.uuid4().hex[:8]}{ext or ".bin"}'
-                p = os.path.join(_TEMP_DIR, fname); open(p, 'wb').write(pt)
+                p = os.path.join(_TEMP_DIR, fname)
+                with open(p, 'wb') as f: f.write(pt)
                 paths.append(p); print(f'[WX] media saved: {fname}', file=sys.__stdout__)
             except Exception as e:
                 print(f'[WX] media dl err ({key}): {e}', file=sys.__stdout__)

ga.py

@@ -20,7 +20,8 @@ def code_run(code, code_type="python", timeout=60, cwd=None, code_cwd=None, stop
     if code_type == "python":
         tmp_file = tempfile.NamedTemporaryFile(suffix=".ai.py", delete=False, mode='w', encoding='utf-8', dir=code_cwd)
         cr_header = os.path.join(script_dir, 'assets', 'code_run_header.py')
-        if os.path.exists(cr_header): tmp_file.write(open(cr_header, encoding='utf-8').read())
+        if os.path.exists(cr_header):
+            with open(cr_header, encoding='utf-8') as _h: tmp_file.write(_h.read())
         tmp_file.write(code)
         tmp_path = tmp_file.name
         tmp_file.close()
@@ -384,8 +385,10 @@ def extract_robust_content(text):
         try:
             new_content = expand_file_refs(blocks, base_dir=self.cwd)
             if mode == "prepend":
-                old = open(path, 'r', encoding="utf-8").read() if os.path.exists(path) else ""
-                open(path, 'w', encoding="utf-8").write(new_content + old)
+                if os.path.exists(path):
+                    with open(path, 'r', encoding="utf-8") as f: old = f.read()
+                else: old = ""
+                with open(path, 'w', encoding="utf-8") as f: f.write(new_content + old)
             else:
                 with open(path, 'a' if mode == "append" else 'w', encoding="utf-8") as f: f.write(new_content)
             yield f"[Status] ✅ {mode.capitalize()} 成功 ({len(new_content)} bytes)\n"
@@ -421,7 +424,8 @@ def enter_plan_mode(self, plan_path):
         print(f"[Info] Entered plan mode with plan file: {plan_path}"); return plan_path
     def _check_plan_completion(self):
         if not os.path.isfile(p:=self._in_plan_mode() or ''): return None
-        try: return len(re.findall(r'\[ \]', open(p, encoding='utf-8', errors='replace').read()))
+        try:
+            with open(p, encoding='utf-8', errors='replace') as f: return len(re.findall(r'\[ \]', f.read()))
         except: return None
 
     def do_update_working_checkpoint(self, args, response):