Run via jcode debug socket using direct OAuth.
(prompt: str)
| 130 | |
| 131 | |
| 132 | def run_jcode_oauth(prompt: str) -> dict: |
| 133 | """Run via jcode debug socket using direct OAuth.""" |
| 134 | print(f"\n{'='*60}") |
| 135 | print("Testing jcode direct OAuth API...") |
| 136 | print(f"{'='*60}") |
| 137 | |
| 138 | # Check if debug socket exists |
| 139 | if not os.path.exists(DEBUG_SOCKET): |
| 140 | return {"error": f"Debug socket not found: {DEBUG_SOCKET}"} |
| 141 | |
| 142 | try: |
| 143 | # Connect to debug socket |
| 144 | sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) |
| 145 | sock.connect(DEBUG_SOCKET) |
| 146 | |
| 147 | # Create a test session |
| 148 | ok, output, err = send_debug_cmd(sock, "create_session:/tmp/oauth-test") |
| 149 | if not ok: |
| 150 | return {"error": f"Failed to create session: {err}"} |
| 151 | |
| 152 | session_data = json.loads(output) |
| 153 | session_id = session_data.get("session_id") |
| 154 | print(f"Created session: {session_id}") |
| 155 | |
| 156 | # Get initial state to confirm provider |
| 157 | ok, output, _ = send_debug_cmd(sock, "state", session_id) |
| 158 | if ok: |
| 159 | state = json.loads(output) |
| 160 | print(f"Provider: {state.get('provider', 'unknown')}") |
| 161 | print(f"Model: {state.get('model', 'unknown')}") |
| 162 | |
| 163 | # Send the test message |
| 164 | start = time.time() |
| 165 | ok, output, err = send_debug_cmd(sock, f"message:{prompt}", session_id, timeout=120) |
| 166 | elapsed = time.time() - start |
| 167 | |
| 168 | print(f"Time: {elapsed:.2f}s") |
| 169 | |
| 170 | if not ok: |
| 171 | send_debug_cmd(sock, f"destroy_session:{session_id}") |
| 172 | sock.close() |
| 173 | return {"error": f"Message failed: {err}", "time": elapsed} |
| 174 | |
| 175 | # The message command returns the text response directly (not JSON) |
| 176 | print(f"Response: {output[:200]}") |
| 177 | |
| 178 | # Query usage via the "usage" command |
| 179 | ok, usage_output, _ = send_debug_cmd(sock, "usage", session_id) |
| 180 | usage = {} |
| 181 | if ok: |
| 182 | try: |
| 183 | usage = json.loads(usage_output) |
| 184 | print(f"\nUsage details:") |
| 185 | print(f" Input tokens: {usage.get('input_tokens', 'N/A')}") |
| 186 | print(f" Output tokens: {usage.get('output_tokens', 'N/A')}") |
| 187 | print(f" Cache read: {usage.get('cache_read_input_tokens', 0) or 0}") |
| 188 | print(f" Cache creation: {usage.get('cache_creation_input_tokens', 0) or 0}") |
| 189 | except json.JSONDecodeError: |
no test coverage detected