#!/usr/bin/env python3 """Persistent local queue daemon for MichelBridge. Runs inside the interactive Windows desktop session. WinRM only drops JSON jobs into C:\MCSJAgent\MichelBridge\queue\in and reads results from queue\out. """ import json, os, re, sys, time, traceback from pathlib import Path ROOT = Path(r"C:\MCSJAgent\MichelBridge") QUEUE_IN = ROOT / "queue" / "in" QUEUE_DONE = ROOT / "queue" / "done" QUEUE_OUT = ROOT / "queue" / "out" QUEUE_BAD = ROOT / "queue" / "bad" LOG = ROOT / "michel_daemon.log" PROOF = Path(r"C:\MCSJAgent\proof\MichelBridge") BANNED_PO_RE = re.compile(r"\b(test|demo|sample|automation|macro|bot|llm|michel|speed\s*test)\b|MICHEL_|_TEST_", re.I) # Import existing bridge primitives from same directory. sys.path.insert(0, str(ROOT)) import michel_bridge as bridge # noqa: E402 def log(msg): ROOT.mkdir(parents=True, exist_ok=True) with LOG.open("a", encoding="utf-8") as f: f.write(time.strftime("%Y-%m-%dT%H:%M:%S%z") + " " + msg + "\n") def atomic_write_json(path: Path, obj): path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(obj, ensure_ascii=False, separators=(",", ":")), encoding="utf-8") os.replace(str(tmp), str(path)) def read_job(path: Path): txt = path.read_text(encoding="utf-8") return json.loads(txt) def safe_po_text_check(job): """Reject internal/test wording in all PO business-facing fields.""" fields = [] for key in ("description", "comment", "header_comment"): if job.get(key): fields.append((key, str(job[key]))) for i, line in enumerate(job.get("lines") or []): if isinstance(line, dict): for key in ("description", "note", "comment"): if line.get(key): fields.append((f"lines[{i}].{key}", str(line[key]))) bad = [] for name, text in fields: m = BANNED_PO_RE.search(text) if m: bad.append({"field": name, "match": m.group(0), "text": text}) if bad: raise ValueError("blocked internal/test wording in PO fields: " + json.dumps(bad, ensure_ascii=False)) def proof_name(prefix): PROOF.mkdir(parents=True, exist_ok=True) return str(PROOF / f"daemon-{prefix}-{time.strftime('%Y%m%d-%H%M%S')}.png") def require_business_reasoning(job, action_name): """Business/write commands must carry a planning envelope so the daemon never blindly clicks.""" if not str(job.get("intent", "")).strip(): raise ValueError(f"{action_name} requires intent: what MCSJ task is being performed and why") if action_name.startswith("po_") and not job.get("approved_training_write"): raise ValueError(f"{action_name} requires approved_training_write=true for demo/training writes") return { "intent": job.get("intent"), "expected_screen": job.get("expected_screen", "Purchase Order Maintenance"), "approved_training_write": bool(job.get("approved_training_write")), } def po_create_plan(job): safe_po_text_check(job) lines = job.get("lines") or [] return { "action": "po_create", "reasoning_contract": "plan-before-execute: validate intent, screen, content, account, then perform deterministic UI steps with proof", "intent": job.get("intent"), "expected_screen": job.get("expected_screen", "Purchase Order Maintenance"), "preflight": [ "activate MCSJ and verify current database/window", "reject internal/test wording in header/comment/line fields", "create header with explicit vendor/description/comment", "open line items and select account by OCR-visible exact account text, not scroll position", "verify selected account crop before saving first line", "save each line, close line window, OCR final header and banned words", ], "vendor": job.get("vendor", "SACCH005"), "description": job.get("description"), "line_count": len(lines), "account_code": job.get("account_code") or (lines[0].get("account") if lines and isinstance(lines[0], dict) else None), } def cmd_plan_po_create(job): return {"ok": True, "plan": po_create_plan(job)} def cmd_health(job): return { "ok": True, "daemon": "michel_daemon", "version": 1, "time": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "pid": os.getpid(), "queue_in": str(QUEUE_IN), "state": bridge.window_state(), } def cmd_activate(job): return {"ok": True, "activate": bridge.activate_mcsj(), "fingerprint": bridge.fingerprint()} def cmd_screenshot(job): return {"ok": True, "screenshot": bridge.screenshot(job.get("path"))} def cmd_ocr_crop(job): return {"ok": True, "ocr": bridge.crop_ocr(int(job["x"]), int(job["y"]), int(job["w"]), int(job["h"]), job.get("path"), int(job.get("psm", 7)))} def cmd_fingerprint(job): return {"ok": True, "fingerprint": bridge.fingerprint()} def cmd_click(job): """Low-level click for calibration only; not for business workflows.""" bridge.activate_mcsj() click(int(job["x"]), int(job["y"]), float(job.get("wait", 0.25))) return {"ok": True, "clicked": [int(job["x"]), int(job["y"])]} def cmd_press(job): bridge.activate_mcsj() key = str(job["key"]) count = int(job.get("count", 1)) for _ in range(count): bridge.tap(key) time.sleep(float(job.get("delay", 0.04))) return {"ok": True, "pressed": key, "count": count} def cmd_account_picklist_probe(job): """Open the PO line-item account picklist from current line screen, OCR it, then Escape.""" bridge.activate_mcsj() before = bridge.screenshot(proof_name("account-probe-before")) click(int(job.get("x", 675)), int(job.get("y", 493)), float(job.get("wait", 0.8))) opened = bridge.screenshot(proof_name("account-probe-opened")) ocr = bridge.crop_ocr(250, 120, 950, 620, proof_name("account-probe-crop"), psm=int(job.get("psm", 6))) bridge.tap('ESC') time.sleep(0.25) after = bridge.screenshot(proof_name("account-probe-after-esc")) return {"ok": True, "before": before, "opened": opened, "ocr": ocr, "after": after} def cmd_po_verify_current(job): """Compact proof/OCR for the currently visible PO header/line screen.""" bridge.activate_mcsj() shot = bridge.screenshot(proof_name("po-current")) # Header area crop works for Purchase Order Maintenance at 1600x1200. header = bridge.crop_ocr(40, 260, 900, 420, proof_name("po-header-crop"), psm=int(job.get("psm", 6))) text = header.get("text", "") bad = sorted(set(m.group(0) for m in BANNED_PO_RE.finditer(text))) return { "ok": True, "screen": "po_verify_current", "screenshot": shot, "header_ocr": header, "banned_matches": bad, } def click(x, y, wait=0.25): import pyautogui pyautogui.click(x, y) time.sleep(wait) def paste_text(text, wait=0.2): import pyautogui, pyperclip pyperclip.copy(text) pyautogui.hotkey("ctrl", "v") time.sleep(wait) def select_all_paste(text, wait=0.2): import pyautogui pyautogui.hotkey("ctrl", "a") time.sleep(0.05) paste_text(text, wait) def save_yes(wait=1.0): click(130, 225, 0.65) # Save click(482, 620, wait) # Yes confirmation def normalize_account_text(s): return re.sub(r"[^A-Z0-9]", "", (s or "").upper()) def _ocr_tokens_for_box(image_path, crop_box, psm=6): from PIL import Image, ImageOps, ImageFilter import pytesseract if os.path.exists(bridge.TESSERACT): pytesseract.pytesseract.tesseract_cmd = bridge.TESSERACT im = Image.open(image_path) x, y, w, h = crop_box crop = im.crop((x, y, x + w, y + h)) gray = ImageOps.grayscale(crop).filter(ImageFilter.SHARPEN) crop_path = proof_name("ocr-crop") gray.save(crop_path) data = pytesseract.image_to_data(gray, config=f"--psm {psm}", output_type=pytesseract.Output.DICT) tokens = [] for i, word in enumerate(data.get("text", [])): word = (word or "").strip() if not word: continue tokens.append({ "text": word, "norm": normalize_account_text(word), "left": int(data["left"][i]) + x, "top": int(data["top"][i]) + y, "width": int(data["width"][i]), "height": int(data["height"][i]), "conf": data.get("conf", [None])[i], }) return crop_path, tokens def _group_tokens_by_row(tokens, tolerance=9): rows = [] for tok in sorted(tokens, key=lambda t: (t["top"] + t["height"] // 2, t["left"])): cy = tok["top"] + tok["height"] // 2 if not rows or abs(rows[-1]["cy"] - cy) > tolerance: rows.append({"cy": cy, "tokens": [tok]}) else: rows[-1]["tokens"].append(tok) rows[-1]["cy"] = int(sum(t["top"] + t["height"] // 2 for t in rows[-1]["tokens"]) / len(rows[-1]["tokens"])) for row in rows: row["text"] = " ".join(t["text"] for t in sorted(row["tokens"], key=lambda t: t["left"])) row["norm"] = normalize_account_text(row["text"]) return rows def _account_code_like_match(row_norm, target_norm): if not target_norm: return False if target_norm in row_norm: return True # OCR often drops the final 0 in 0400. if len(target_norm) >= 8 and target_norm[:-1] in row_norm: return True m = re.search(r"A\d{7,8}", row_norm) if not m: return False row_code = m.group(0)[:len(target_norm)] if len(row_code) != len(target_norm): return False # Allow one OCR digit error (e.g. 1110 read as 1118), but not a different account family. dist = sum(1 for a, b in zip(row_code, target_norm) if a != b) return dist <= 1 def _account_row_matches(row_norm, target_norm, label_norm=""): code_ok = _account_code_like_match(row_norm, target_norm) if not label_norm: return code_ok # Label is a secondary guard, never a substitute for a conflicting account code. return code_ok and label_norm in row_norm def ocr_find_account_in_region(account_code, account_label=None, crop_box=(430, 260, 730, 450)): """Return screen click point for an OCR-visible account row in the open picklist. Hard rule: choose the row whose OCR text matches the requested account/label. Do not scroll-position-click. This prevents selecting A-1110-0000 when A-1110-0400 was intended. """ shot = bridge.screenshot(proof_name("account-picklist-screen")) crop_path, tokens = _ocr_tokens_for_box(shot["path"], crop_box, psm=6) rows = _group_tokens_by_row(tokens) target = normalize_account_text(account_code) label_norm = normalize_account_text(account_label or "") candidates = [] for row in rows: if _account_row_matches(row["norm"], target, label_norm): # Click in the row's account-code/text area, not on a left tree control. candidates.append({"cy": row["cy"], "text": row["text"], "norm": row["norm"], "click": [760, row["cy"]]}) if not candidates: return {"found": False, "target": target, "label": label_norm, "rows": rows[:30], "proof_crop": crop_path, "screen": shot} # Prefer the first row whose account prefix is close to the target and whose text is not a rollup-only 0000 when target is not 0000. def score(c): penalty = 0 if target.endswith("0400") and "0000" in c["norm"]: penalty += 100 if label_norm and label_norm in c["norm"]: penalty -= 10 if target in c["norm"]: penalty -= 20 return penalty best = sorted(candidates, key=score)[0] return {"found": True, "click": best["click"], "target": target, "label": label_norm, "matched": best, "candidates": candidates, "proof_crop": crop_path, "screen": shot} def detect_control_account_message(): shot = bridge.screenshot(proof_name("account-message-check")) crop_path, tokens = _ocr_tokens_for_box(shot["path"], (560, 560, 520, 210), psm=6) text = " ".join(t["text"] for t in tokens) norm = normalize_account_text(text) return {"screen": shot, "crop": crop_path, "text": text, "norm": norm, "is_control_message": "ACCOUNTENTEREDISACONTROLACCOUNT" in norm or ("CONTROLACCOUNT" in norm and "MESSAGE" in norm)} def _find_account_visible_with_scroll(account_code, account_label=None, max_pages=8): import pyautogui attempts = [] # Normalize to top first so row positions are deterministic across previous picks. for key in ("home", None): if key: pyautogui.press(key) time.sleep(0.35) for page in range(max_pages + 1): found = ocr_find_account_in_region(account_code, account_label=account_label) attempts.append(found) if found.get("found"): found["attempt_summaries"] = [ {"found": a.get("found"), "target": a.get("target"), "label": a.get("label"), "matched": a.get("matched"), "rows_count": len(a.get("rows", []))} for a in attempts[-3:] ] found["page_index"] = page return found pyautogui.press("pagedown") time.sleep(0.35) return attempts[-1] if attempts else {"found": False, "error": "no attempts"} def select_account_verified(account_code, account_label=None, ellipsis=(918, 587), ok=(672, 622), selected_crop=(430, 560, 560, 130)): """Open account picklist, find requested account row by OCR, click it, OK, and verify no control-account modal.""" click(int(ellipsis[0]), int(ellipsis[1]), 0.7) found = _find_account_visible_with_scroll(account_code, account_label=account_label) if not found.get("found"): # Try one deterministic filter pass: close picklist, type target account in the # account field, then reopen. If it still is not OCR-visible, abort. bridge.tap('ESC') time.sleep(0.2) field_x = max(500, int(ellipsis[0]) - 70) field_y = int(ellipsis[1]) click(field_x, field_y, 0.08) select_all_paste(str(account_code), 0.15) click(int(ellipsis[0]), int(ellipsis[1]), 0.7) found = _find_account_visible_with_scroll(account_code, account_label=account_label) if not found.get("found"): bridge.tap('ESC') raise ValueError("requested account not OCR-visible in picklist after filter pass; refusing scroll-position fallback: " + json.dumps(found, ensure_ascii=False)[:1600]) click(found["click"][0], found["click"][1], 0.15) click(int(ok[0]), int(ok[1]), 0.65) msg = detect_control_account_message() if msg.get("is_control_message"): # Leave the modal up for caller/proof; do not continue toward save. raise ValueError("account selection produced control-account message; refusing to save: " + json.dumps({"selection": found, "message": msg}, ensure_ascii=False)[:1600]) selected = bridge.crop_ocr(*selected_crop, proof_name("account-selected-crop"), psm=6) selected_norm = normalize_account_text(selected.get("text", "")) target_norm = normalize_account_text(account_code) label_norm = normalize_account_text(account_label or "") code_verified = target_norm in selected_norm or (len(target_norm) >= 8 and target_norm[:-1] in selected_norm) or _account_code_like_match(selected_norm, target_norm) if not code_verified: raise ValueError(f"account verification failed after pick: expected account {account_code}, OCR={selected.get('text')!r}, selection={found}") return {"selection": found, "selected_ocr": selected, "message_check": msg} def fill_po_line(seq, desc, price, account_code="F-1910-0400"): click(60, 225, 0.55) # Add line if seq == 1: click(802, 633, 0.9) # Add Line Item OK; leave duplicate unchecked click(175, 462, 0.1); select_all_paste(desc, 0.15) account_result = select_account_verified(account_code) click(145, 705, 0.1); select_all_paste(price, 0.15) else: click(690, 570, 0.08) # duplicate previous checkbox click(802, 633, 0.75) click(175, 462, 0.1); select_all_paste(desc, 0.15) click(145, 705, 0.1); select_all_paste(price, 0.15) account_result = {"selection": "duplicated_from_previous_verified_line"} save_yes(1.0) return account_result def cmd_po_account_select_dryrun(job): """Dry-run account OCR selection from current unsaved PO line screen. Does not save.""" bridge.activate_mcsj() result = select_account_verified( str(job.get("account") or "A-1010-0400"), account_label=str(job.get("account_label") or "CONTRACTUAL EXPENSE"), ellipsis=tuple(job.get("ellipsis") or (918, 587)), ok=tuple(job.get("ok") or (672, 622)), selected_crop=tuple(job.get("selected_crop") or (430, 560, 560, 130)), ) shot = bridge.screenshot(proof_name("account-select-dryrun-after")) return {"ok": True, "selection": result, "screenshot": shot} def cmd_po_create_speed_centered(job): """Create a simple one-line PO from a visible centered Purchase Order Maintenance header. Verified start map: 1600x1200, internal Purchase Order Maintenance centered or already maximized, header saved/not in edit mode. Account selection is OCR row-verified before OK and aborts on control-account message. """ reasoning = require_business_reasoning(job, "po_create_speed_centered") safe_po_text_check(job) lines = job.get("lines") or [] if len(lines) != 1: raise ValueError("po_create_speed_centered requires exactly one line") line = lines[0] desc = str(job["description"]) line_desc = str(line["description"]) price = str(line.get("price", "1.00")) vendor = str(job.get("vendor", "10T07005")) start = time.perf_counter() proof = {} bridge.activate_mcsj() proof["before"] = bridge.screenshot(proof_name("speed-before")) start_title = bridge.crop_ocr(360, 315, 900, 80, proof_name("speed-start-title"), psm=6) start_norm = normalize_account_text(start_title.get("text", "")) if "LINEITEMMAINTENANCE" in start_norm or "PURCHASEORDERMAINTENANCE" not in start_norm: raise ValueError(f"unsafe start screen for po_create_speed_centered; expected Purchase Order Maintenance header, OCR={start_title.get('text')!r}") proof["start_title"] = start_title # Starting point map: centered Purchase Order Maintenance header. click(1238, 337, 0.45) # maximize internal PO header if centered proof["header_maximized"] = bridge.screenshot(proof_name("speed-header-max")) click(56, 100, 0.8) # Add header bridge.tap('ENTER'); time.sleep(0.55) # Select Prefix/Add Record defaults bridge.tap('ENTER'); time.sleep(1.0) proof["add_mode"] = bridge.screenshot(proof_name("speed-add-mode")) click(496, 166, 0.08); paste_text(vendor, 0.15) click(176, 256, 0.08); paste_text(desc, 0.15) # Force valid vendor population via picklist; typed id alone can fail validation. click(578, 166, 0.55) # vendor ellipsis proof["vendor_picklist"] = bridge.screenshot(proof_name("speed-vendor-picklist")) click(676, 803, 0.75) # OK on visible first/vendor row click(132, 100, 0.65) # Save header click(763, 688, 1.0) # Yes confirmation proof["header_saved"] = bridge.screenshot(proof_name("speed-header-saved")) click(647, 100, 0.9) # Line Item from maximized header proof["line_open"] = bridge.screenshot(proof_name("speed-line-open")) # Line-item centered map. click(413, 366, 0.7) # line Add click(776, 625, 0.75) # Add Line Item OK click(575, 561, 0.08); paste_text(line_desc, 0.15) account_result = select_account_verified( str(line.get("account") or job.get("account_code") or "A-1010-0400"), account_label=str(line.get("account_label") or "CONTRACTUAL EXPENSE"), ellipsis=(918, 587), ok=(672, 622), selected_crop=(430, 560, 560, 130), ) proof["account_selected"] = account_result click(543, 778, 0.08); select_all_paste(price, 0.15) proof["line_filled"] = bridge.screenshot(proof_name("speed-line-filled")) click(489, 366, 0.65) # Save line click(742, 700, 0.8) # Yes save confirmation # Ordinary sub-account over-encumbrance is expected in demo DB for this account/amount. click(874, 711, 0.9) # Yes warning if present; harmless click if already dismissed area proof["final_line"] = bridge.screenshot(proof_name("speed-final-line")) verify = cmd_po_verify_current({"psm": 6}) return {"ok": True, "reasoning": reasoning, "elapsed_seconds": round(time.perf_counter() - start, 2), "description": desc, "line": line, "vendor": vendor, "proof": proof, "verify": verify} def cmd_po_create(job): """Create a PO using the currently mapped PO Maintenance path. Requires current visible screen to be Purchase Order Maintenance. This is a demo/training command and uses calibrated 1600x1200 coordinates plus OCR-verified account selection. """ reasoning = require_business_reasoning(job, "po_create") plan = po_create_plan(job) vendor = job.get("vendor", "SACCH005") desc = job["description"] comment = job.get("comment") or job.get("header_comment") or "" lines = job.get("lines") or [] if not lines: raise ValueError("po_create requires at least one line") if len(lines) > 10: raise ValueError("po_create line limit is 10 for this mapped command") import pyautogui pyautogui.FAILSAFE = False pyautogui.PAUSE = 0.05 start = time.perf_counter() bridge.activate_mcsj() proof = {"before": bridge.screenshot(proof_name("po-create-before"))} click(60, 235, 0.65) # Add header click(522, 586, 0.65) # Select Prefix OK click(560, 592, 1.1) # Add Record OK click(600, 315, 0.1); paste_text(vendor, 0.25) click(185, 405, 0.1); paste_text(desc, 0.25) if comment: click(660, 600, 0.1); paste_text(comment, 0.25) save_yes(1.25) proof["header_saved"] = bridge.screenshot(proof_name("po-create-header-saved")) click(755, 225, 0.9) # LineItem account_results = [] account_code = job.get("account_code") or "F-1910-0400" for idx, line in enumerate(lines, 1): line_account = str(line.get("account") or account_code) account_results.append(fill_po_line(idx, str(line["description"]), str(line.get("price", "0.01")), line_account)) proof["lines_saved"] = bridge.screenshot(proof_name("po-create-lines-saved")) click(250, 225, 0.8) # Close line item window proof["final_header"] = bridge.screenshot(proof_name("po-create-final-header")) verify = cmd_po_verify_current({"psm": 6}) return {"ok": True, "reasoning": reasoning, "plan": plan, "elapsed_seconds": round(time.perf_counter() - start, 2), "description": desc, "comment": comment, "line_count_requested": len(lines), "account_results": account_results, "proof": proof, "verify": verify} COMMANDS = { "health": cmd_health, "activate": cmd_activate, "screenshot": cmd_screenshot, "ocr_crop": cmd_ocr_crop, "fingerprint": cmd_fingerprint, "plan_po_create": cmd_plan_po_create, "click": cmd_click, "press": cmd_press, "account_picklist_probe": cmd_account_picklist_probe, "po_account_select_dryrun": cmd_po_account_select_dryrun, "po_verify_current": cmd_po_verify_current, "po_create_speed_centered": cmd_po_create_speed_centered, "po_create": cmd_po_create, } def handle_file(path: Path): job_id = path.stem started = time.perf_counter() try: job = read_job(path) action = job.get("action") if action not in COMMANDS: raise ValueError(f"unknown action: {action}") result = COMMANDS[action](job) result.setdefault("ok", True) result["job_id"] = job.get("id") or job_id result["action"] = action result["elapsed_seconds_total"] = round(time.perf_counter() - started, 3) atomic_write_json(QUEUE_OUT / f"{job_id}.json", result) path.replace(QUEUE_DONE / path.name) log(f"DONE {job_id} {action} {result['elapsed_seconds_total']}s") except Exception as e: err = { "ok": False, "job_id": job_id, "error": str(e), "traceback": traceback.format_exc(limit=20), "elapsed_seconds_total": round(time.perf_counter() - started, 3), } atomic_write_json(QUEUE_OUT / f"{job_id}.json", err) try: path.replace(QUEUE_BAD / path.name) except Exception: pass log(f"ERROR {job_id} {e}") def main(): for d in (QUEUE_IN, QUEUE_DONE, QUEUE_OUT, QUEUE_BAD, PROOF): d.mkdir(parents=True, exist_ok=True) log("START daemon pid=" + str(os.getpid())) # Mark ready for cheap WinRM health checks. atomic_write_json(ROOT / "michel_daemon.status.json", {"ok": True, "pid": os.getpid(), "started": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "queue_in": str(QUEUE_IN), "queue_out": str(QUEUE_OUT)}) while True: jobs = sorted(QUEUE_IN.glob("*.json"), key=lambda p: p.stat().st_mtime) if not jobs: time.sleep(0.2) continue for path in jobs: handle_file(path) if __name__ == "__main__": main()