Variet/gravity_control
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: subscribeToStream on new cascade — real-time StreamCascadeReactiveUpdates subscription

Browse Source
This commit is contained in:
authentik Default Admin
2026-03-07 23:23:16 +09:00
parent 12131b9103
commit 87094e00b0
4 changed files with 124 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
extension/src/extension.ts
View File

@@ -313,7 +313,7 @@ export function activate(context: vscode.ExtensionContext) {
} else if (stepIdx > 0 && summary) {
// New conversation with AI response!
console.log(`Gravity Bridge: [LS] NEW conversation ${key.substring(0, 8)} at step ${stepIdx} "${summary.substring(0, 40)}"`);
subscribeToStream(key);
// Try to extract AI text from extensionLogs
const aiText = extractFromLogs(parsed);
if (aiText) {
@@ -429,95 +429,74 @@ export function activate(context: vscode.ExtensionContext) {
// ========== Trial D2: Streaming RPC with protocol_version=1 ==========
setTimeout(async () => {
if (!lsPort || !lsCsrf) { console.log('Gravity Bridge: [Trial D2] Skipped — no LS'); return; }
console.log(`Gravity Bridge: [Trial D2] Streaming RPCs with proto_version=1...`);
// ========== Stream subscription for active cascades ==========
function subscribeToStream(cascadeId: string): void {
if (!lsPort || !lsCsrf) { return; }
const http = require('http');
console.log(`Gravity Bridge: [Stream] Subscribing to cascade ${cascadeId.substring(0, 8)}...`);
function tryProtoRPC(method: string, protoBody: Buffer, timeout: number = 12000): Promise<string> {
return new Promise((resolve) => {
// ConnectRPC frame: [flag(1)] [length(4 big-endian)] [message]
const frame = Buffer.alloc(5 + protoBody.length);
frame[0] = 0x00; // no compression
frame.writeUInt32BE(protoBody.length, 1);
protoBody.copy(frame, 5);
const gidBuf = Buffer.from(cascadeId, 'utf-8');
// field1=version(varint=1), field2=cascadeId(string)
const proto = Buffer.alloc(2 + 2 + gidBuf.length);
proto[0] = 0x08; proto[1] = 0x01;
proto[2] = 0x12; proto[3] = gidBuf.length;
gidBuf.copy(proto, 4);
const req = http.request({
hostname: '127.0.0.1', port: lsPort,
path: `/exa.language_server_pb.LanguageServerService/${method}`,
method: 'POST',
headers: {
'Content-Type': 'application/connect+proto',
'Connect-Protocol-Version': '1',
'x-codeium-csrf-token': lsCsrf,
},
timeout: timeout
}, (res: any) => {
const chunks: Buffer[] = [];
res.on('data', (c: Buffer) => chunks.push(c));
res.on('end', () => {
const buf = Buffer.concat(chunks);
// Skip ConnectRPC frame header (5 bytes) to get proto/JSON content
const text = buf.toString('utf-8');
resolve(text);
});
});
req.on('error', (e: any) => resolve(`err:${e.message}`));
req.on('timeout', () => { req.destroy(); resolve('timeout(stream open)'); });
req.write(frame); req.end();
// ConnectRPC frame
const frame = Buffer.alloc(5 + proto.length);
frame[0] = 0x00;
frame.writeUInt32BE(proto.length, 1);
proto.copy(frame, 5);
const req = http.request({
hostname: '127.0.0.1', port: lsPort,
path: '/exa.language_server_pb.LanguageServerService/StreamCascadeReactiveUpdates',
method: 'POST',
headers: {
'Content-Type': 'application/connect+proto',
'Connect-Protocol-Version': '1',
'x-codeium-csrf-token': lsCsrf,
},
timeout: 30000
}, (res: any) => {
console.log(`Gravity Bridge: [Stream] Connected! status=${res.statusCode}`);
let totalBytes = 0;
const allChunks: Buffer[] = [];
res.on('data', (chunk: Buffer) => {
allChunks.push(chunk);
totalBytes += chunk.length;
// Log each chunk as it arrives (real-time streaming)
const text = chunk.toString('utf-8');
console.log(`Gravity Bridge: [Stream] chunk (${chunk.length}B): ${text.substring(0, 300)}`);
});
}
// Get latest trajectory ID first, then try streams WITH cascadeId
try {
const dRaw = await vscode.commands.executeCommand('antigravity.getDiagnostics');
const d = typeof dRaw === 'string' ? JSON.parse(dRaw) : dRaw;
const ts = d?.recentTrajectories || [];
if (ts.length > 0) {
const latest = ts[ts.length - 1];
const gid = latest.googleAgentId || '';
console.log(`Gravity Bridge: [Trial D3] Using cascade ${gid.substring(0, 8)} step=${latest.lastStepIndex}`);
res.on('end', () => {
const fullBuf = Buffer.concat(allChunks);
const fullText = fullBuf.toString('utf-8');
console.log(`Gravity Bridge: [Stream] Ended. Total ${totalBytes}B in ${allChunks.length} chunks`);
const gidBuf = Buffer.from(gid, 'utf-8');
// Try to extract readable text from the stream data
// Look for strings in the protobuf data
const readable = fullText.replace(/[\x00-\x1F\x7F-\x9F]/g, ' ').replace(/\s+/g, ' ').trim();
if (readable.length > 20) {
console.log(`Gravity Bridge: [Stream] Readable: ${readable.substring(0, 1000)}`);
}
// Attempt 1: field1=version(varint=1), field2=cascadeId(string)
// 0x08 0x01 0x12 <len> <cascadeId>
const proto1 = Buffer.alloc(2 + 2 + gidBuf.length);
proto1[0] = 0x08; proto1[1] = 0x01; // field1 varint = 1
proto1[2] = 0x12; proto1[3] = gidBuf.length; // field2 string
gidBuf.copy(proto1, 4);
let r = await tryProtoRPC('StreamCascadeReactiveUpdates', proto1);
console.log(`Gravity Bridge: [Trial D3] StreamCascade(v1+cascade): ${r.substring(0, 800)}`);
// Attempt 2: field1=cascadeId(string), field2=version(varint=1)
// 0x0A <len> <cascadeId> 0x10 0x01
const proto2 = Buffer.alloc(2 + gidBuf.length + 2);
proto2[0] = 0x0A; proto2[1] = gidBuf.length; // field1 string
gidBuf.copy(proto2, 2);
proto2[2 + gidBuf.length] = 0x10; // field2 varint
proto2[3 + gidBuf.length] = 0x01;
r = await tryProtoRPC('StreamCascadeReactiveUpdates', proto2);
console.log(`Gravity Bridge: [Trial D3] StreamCascade(cascade+v1): ${r.substring(0, 800)}`);
// Attempt 3: StreamCascadeSummariesReactiveUpdates with version+cascadeId
r = await tryProtoRPC('StreamCascadeSummariesReactiveUpdates', proto1);
console.log(`Gravity Bridge: [Trial D3] StreamSummaries(v1+cascade): ${r.substring(0, 800)}`);
// Attempt 4: GetCascadeTrajectorySteps with proto
const stepIdx = Math.max(0, latest.lastStepIndex - 1);
const stepsProto = Buffer.alloc(2 + gidBuf.length + 2);
stepsProto[0] = 0x0A; stepsProto[1] = gidBuf.length;
gidBuf.copy(stepsProto, 2);
stepsProto[2 + gidBuf.length] = 0x10;
stepsProto[3 + gidBuf.length] = stepIdx;
r = await tryProtoRPC('GetCascadeTrajectorySteps', stepsProto);
console.log(`Gravity Bridge: [Trial D3] Steps(proto): ${r.substring(0, 1000)}`);
}
} catch (e: any) { console.log(`Gravity Bridge: [Trial D3] err: ${e.message}`); }
}, 15000);
// Check for error messages
if (fullText.includes('"error"')) {
console.log(`Gravity Bridge: [Stream] Error in response: ${fullText.substring(0, 500)}`);
}
});
});
req.on('error', (e: any) => console.log(`Gravity Bridge: [Stream] Error: ${e.message}`));
req.on('timeout', () => {
console.log(`Gravity Bridge: [Stream] Timeout (30s) — closing`);
req.destroy();
});
req.write(frame);
req.end();
}
// Start LS bridge after a delay